Distributed Job Scheduler ডিজাইন

ধরো তোমার platform-এ লাখ লাখ কাজ নির্দিষ্ট সময়ে চলতে হয় — প্রতি রাত ২টায় report বানানো, ৩০ মিনিট পর payment reminder পাঠানো, প্রতি ঘণ্টায় cache refresh। একটা single-server cron দিয়ে এটা চলবে না — server crash করলে job মিস হবে, scale বাড়লে একটা machine কুলোবে না। তাই দরকার একটা distributed job scheduler। চলো ডিজাইন করি।

১. সমস্যা বোঝা (Requirements)

Functional requirements:

• One-time job schedule করা (যেমন "১০ মিনিট পর এই email পাঠাও")।
• Recurring / cron job schedule করা (যেমন "প্রতিদিন রাত ২টায়")।
• নির্দিষ্ট সময়ে job execute করা।
• ব্যর্থ হলে retry করা, নির্দিষ্ট policy অনুযায়ী।
• Job-এর status track করা (pending, running, succeeded, failed)।

Non-functional requirements:

• At-least-once execution — job মিস হওয়া যাবে না।
• No double-run (বা double-run হলেও নিরাপদ) — একই job দুইবার যেন ক্ষতি না করে।
• Scalability — millions of jobs।
• Fault tolerance — কোনো worker/scheduler crash করলেও system চলবে।
• Reasonable timing accuracy — কয়েক সেকেন্ডের মধ্যে trigger।

২. স্কেল আন্দাজ (Estimation)

ধরি: 10 million scheduled job/day
=> গড় = 10,000,000 / 86,400 ≈ 116 job/s
peak (গড়ের 5x) ≈ 580 job/s

কিন্তু ঘড়ির কাঁটার মাথায় (যেমন রাত ২:০০:০০) thundering herd:
ধরা যাক 500,000 job একই মিনিটে due
=> 500,000 / 60 ≈ 8,300 job/s সেই মিনিটে

Storage:
প্রতি job metadata ~1 KB, 30 দিন রাখলে
10M * 30 * 1 KB = ~300 GB

Worker:
একটি worker ~50 job/s চালালে peak-এ
8,300 / 50 ≈ 170 worker দরকার

মূল শিক্ষা: জটিলতা throughput-এ নয়, বরং সময়ের spike (সবাই একই মুহূর্তে due) আর exactly সঠিক সময়ে trigger করায়।

৩. API ডিজাইন

scheduleJob(payload, runAt, options) -> jobId
  - runAt: নির্দিষ্ট timestamp (one-time)
  - options: {maxRetries, backoff, idempotencyKey, priority}

scheduleCron(payload, cronExpr, options) -> jobId
  - cronExpr: "0 2 * * *" (প্রতিদিন রাত ২টা)

cancelJob(jobId) -> ok
getJobStatus(jobId) -> {state, attempts, lastError, nextRunAt}

লক্ষ্য করো idempotencyKey — caller চাইলে নিজের key দিতে পারে, যাতে একই কাজ দুবার schedule হলেও একবারই effect হয়।

৪. ডেটা মডেল

Field	বর্ণনা
jobId	unique id
payload	কী execute হবে (handler + args)
nextRunAt	পরবর্তী কখন due (indexed)
state	PENDING / RUNNING / SUCCEEDED / FAILED / DEAD
attempts	কতবার চেষ্টা হয়েছে
cronExpr	recurring হলে
lockedBy / lockExpiry	কোন worker ধরেছে, কখন lock শেষ
idempotencyKey	duplicate ঠেকাতে

Storage choice: আমরা একটা database নিচ্ছি যেখানে nextRunAt-এ index আছে (যেমন PostgreSQL বা একটা sharded NoSQL)। কারণ scheduler-এর core query হলো "এখন বা তার আগে due এমন job গুলো দাও" — এটা WHERE nextRunAt এর কম now AND state = PENDING ধাঁচের range scan, যার জন্য sorted index আদর্শ।

একটা option হলো time-bucketed approach — job গুলোকে minute-bucket-এ ভাগ করে রাখা, যাতে প্রতি tick-এ শুধু সংশ্লিষ্ট bucket scan করতে হয়, পুরো table নয়।

৫. হাই-লেভেল ডিজাইন

মূল components:

• API / Ingestion service: job নেয়, validate করে job store-এ লেখে।
• Job store (DB): persistent state ও nextRunAt index।
• Scheduler (poller): নিয়মিত (যেমন প্রতি সেকেন্ডে) due job খোঁজে এবং ready queue-তে দেয়। একাধিক scheduler থাকলে leader election বা partitioning দিয়ে কাজ ভাগ হয়।
• Queue: due job গুলোর জন্য একটা message queue (Kafka/SQS/RabbitMQ)।
• Worker pool: queue থেকে job নিয়ে execute করে, status update করে।
• Retry/DLQ handler: ব্যর্থ job-এর জন্য backoff ও dead-letter সামলায়।

Flow:

1. Client scheduleJob করে → job store-এ PENDING হিসেবে লেখা।
2. Scheduler poll করে nextRunAt এর কম now job বের করে।
3. সেগুলো queue-তে push করে, state RUNNING করে।
4. Worker queue থেকে নেয়, lock নিশ্চিত করে, execute করে।
5. সফল হলে SUCCEEDED; cron হলে পরবর্তী nextRunAt হিসাব করে আবার PENDING।
6. ব্যর্থ হলে retry policy অনুযায়ী backoff-সহ reschedule, অথবা DLQ।

৬. গভীরে (Deep Dive)

ক. Double-run ঠেকানো: Leader Election + Distributed Lock

দুটো level-এ সুরক্ষা লাগে।

Scheduler level: একাধিক scheduler instance চললে সবাই একই due job তুলে নিতে পারে। সমাধান —

• Leader election (ZooKeeper/etcd/Redis-এর lease) দিয়ে এক scheduler-ই poll করবে; অথবা
• Partitioning — job space কে shard করে প্রতিটি scheduler নির্দিষ্ট shard সামলায়।

Job level: worker job তোলার সময় একটা distributed lock নেয়। উদাহরণ — DB-তে atomic conditional update:

UPDATE jobs
SET state = 'RUNNING', lockedBy = 'worker_7', lockExpiry = now + 60s
WHERE jobId = 'X' AND state = 'PENDING';
-- শুধু একটি worker-ই এই row পাবে (rows affected = 1 হলেই lock পেলো)

অথবা Redis-এ SET key worker_id NX EX 60 দিয়ে lock। lockExpiry/TTL জরুরি — worker crash করলে lock চিরকাল আটকে থাকবে না, TTL শেষে অন্য worker নেবে।

খ. Idempotency

Distributed system-এ at-least-once অনিবার্য — lock TTL শেষ হয়ে দুই worker একই job ধরা, বা network glitch-এ message redelivery। তাই lock যথেষ্ট নয়, execution-ও idempotent হতে হবে।

• প্রতিটি execution-এর সাথে একটা executionId বা idempotencyKey থাকবে।
• Worker কাজ করার আগে দেখবে এই key আগে process হয়েছে কিনা (একটা dedup store/table-এ)।
• হয়ে থাকলে skip করবে, না হলে কাজ করে key mark করবে।

যেমন payment job — "একই idempotencyKey-তে দ্বিতীয়বার charge কোরো না" নিশ্চিত করলে double-run হলেও user দুবার টাকা কাটবে না।

গ. Retry, Backoff ও Dead-Letter

ব্যর্থ job সামলানোর policy:

• Exponential backoff with jitter: ১ম retry ১০s পর, ২য় ৩০s, ৩য় ৯০s... সাথে random jitter যাতে সব fail-করা job একসাথে আবার hit না করে (thundering herd এড়াতে)।
• maxRetries: নির্দিষ্ট সংখ্যা পর থামো।
• Dead-letter queue (DLQ): maxRetries-এর পরও fail করলে job DLQ-তে সরাও, state = DEAD। এতে main pipeline পরিষ্কার থাকে, আর team পরে DLQ inspect করে root cause বের করে manually replay করতে পারে।

৭. বটলনেক ও স্কেলিং

• Thundering herd at clock boundary: রাত ২:০০:০০-এ লাখো cron একসাথে due। সমাধান — execution time-এ ছোট jitter যোগ করা, scheduler-এ batch dispatch, আর time-bucket sharding।
• Scheduler poll load: ঘন ঘন full-table scan ব্যয়বহুল। nextRunAt index + bucketed query + leader/partition দিয়ে নিয়ন্ত্রণ করো।
• Hot shard: কিছু shard-এ বেশি job জমলে dynamic rebalancing দরকার।
• DB bottleneck: lock-এর জন্য DB contention হলে Redis-ভিত্তিক lock বা queue-driven dispatch-এ সরানো যায়।
• Long-running job: lock TTL-এর চেয়ে job বেশি সময় নিলে সমস্যা — তাই worker-কে periodically lock renew (heartbeat) করতে হবে।
• Clock skew: distributed node-দের ঘড়ি ভিন্ন হলে timing ভুল হয়; NTP sync ও একটা authoritative time source ব্যবহার করো।

৮. সারসংক্ষেপ

একটা distributed job scheduler-এর মূল চারটি স্তম্ভ: একটা durable job store (nextRunAt-indexed), একটা scheduler যা leader election বা partition দিয়ে নিরাপদে due job তোলে, distributed lock + idempotency-এর দ্বৈত সুরক্ষা যাতে double-run ক্ষতি না করে, আর retry + backoff + DLQ দিয়ে নির্ভরযোগ্য failure handling। সবচেয়ে বড় শিক্ষা — lock দিয়ে double-run কমাও, কিন্তু idempotency দিয়েই সত্যিকার নিরাপত্তা পাও।