BullMQ 6 في NestJS: المهام المتوقفة، ومنع التكرار، ونقل المهام المتكررة إلى Job Schedulers — ملاحظات ميدانية

مكتبة BullMQ، وهي طابور مهام (job queue) لـ Node.js قائم على Redis، تفشل في بيئة الإنتاج غالبًا بسبب الإعدادات لا بسبب الكود. فإعداد maxmemory-policy في Redis بأي قيمة غير noeviction يسمح لـ Redis بحذف مفاتيح الطابور. والمعالج (processor) الذي يستهلك المعالج المركزي ويحجب حلقة الأحداث (event loop) أكثر من مدة قفل المهمة البالغة 30 ثانية يجعل المهمة نفسها تُنفَّذ مرتين. ومعرّف المهمة المخصص jobId يُسقط مهامًا مشروعة دون أي خطأ. كما أن الإصدار BullMQ 6.0.0، الصادر في 2026-07-30، يزيل خيار repeat القديم، لذا يجب نقل المهام المتكررة إلى Job Schedulers قبل الترقية.

في Devya نشغّل BullMQ خلف واجهات NestJS البرمجية لتنفيذ الأعمال التي لا ينبغي أن تجري داخل طلب HTTP: رسائل البريد الإلكتروني للمعاملات، وتقارير PDF، وتوزيع أحداث webhooks، ومهام التنظيف الليلية. بناء المسار السعيد يستغرق ظهيرة واحدة؛ أما الأعطال فاستغرقت من فريقنا وقتًا أطول، لأن أيًّا منها لم يُطلق خطأً في المكان الذي كنا نراقبه. كُتبت هذه الملاحظات على BullMQ 6.5 و@nestjs/bullmq 12.

أهم النقاط

يجب تشغيل Redis المخصص لـ BullMQ بإعداد maxmemory-policy noeviction. توثيق BullMQ يصفه بأنه الإعداد الوحيد الذي يضمن سلوكًا صحيحًا للطوابير، وهذه السياسة تنطبق على نسخة Redis بالكامل لا على قاعدة بيانات منطقية واحدة.

المهمة المتوقفة (stalled job) هي مهمة انتهت صلاحية قفلها بينما كانت لا تزال قيد التنفيذ. تعيدها BullMQ إلى قائمة الانتظار فينفذها عامل (worker) آخر من جديد، لذا مكان المعالجات كثيفة الاستهلاك للمعالج هو المعالجات المعزولة (sandboxed processors).

تسلّم BullMQ المهام مرة واحدة على الأقل (at-least-once)، لذا يجب أن يكون كل معالج متساوي القوة (idempotent) حتى عند استخدام jobId أو deduplication.

معرّف jobId المخصص يمنع التكرار مقابل كل مهمة لا تزال مخزنة في Redis، بما في ذلك المهام المكتملة التي يحتفظ بها removeOnComplete. أما خيار deduplication فيمنع التكرار فقط ما دامت المهمة غير منتهية أو داخل مدة TTL.

يُطلق BullMQ 6 خطأً عند العثور على بيانات المهام المتكررة القديمة في Redis، لذا ننقل كل مهمة repeat إلى upsertJobScheduler() ونحن لا نزال على BullMQ 5.

كيف نُعدّ BullMQ 6 في تطبيق NestJS؟

إعداد BullMQ للإنتاج في NestJS يتكون من استدعاء واحد لـ BullModule.forRootAsync() لاتصال Redis، واستدعاء BullModule.registerQueue() لكل طابور مع خيارات افتراضية لإعادة المحاولة والاحتفاظ، وصنف معالج واحد لكل طابور يرث من WorkerHost. جعل BullMQ 6 مكتبة ioredis تبعية نظيرة اختيارية (optional peer dependency)، لذا نثبّت bullmq و@nestjs/bullmq وioredis صراحةً.

نضع الإعدادات الافتراضية لإعادة المحاولة على مستوى الطابور لا في كل استدعاء add(): خمس محاولات، وتراجع أُسّي (exponential backoff) يبدأ من ثانيتين، وقيمة jitter تساوي 0.5. ينتظر التراجع الأسي 2^(attempts - 1) × delay ميلي ثانية قبل كل إعادة محاولة، وقيمة jitter 0.5 تجعل كل تأخير عشوائيًا بين نصف تلك القيمة وكاملها، فلا تعيد موجة من الأعطال المحاولة في اللحظة نفسها.

يقبل الخياران removeOnComplete وremoveOnFail عمرًا (age) بالثواني وعددًا (count). نحتفظ بالمهام المكتملة 24 ساعة أو 5,000 مهمة، وبالمهام الفاشلة 7 أيام. وبدون هذين الخيارين يحتفظ BullMQ بكل مهمة منتهية في Redis إلى الأبد.

المعالج مزوّد (provider) عادي في NestJS، لذا يمكنه حقن الخدمات. إطلاق UnrecoverableError من bullmq ينقل المهمة مباشرة إلى مجموعة المهام الفاشلة دون استهلاك محاولاتها المتبقية، وهو بديل Job#discard() الذي أزاله BullMQ 6. نستخدمه للأخطاء التي لن تُصلح نفسها، مثل عنوان مستلم غير صالح.

لماذا يفقد BullMQ المهام عندما يُستخدم Redis كذاكرة تخزين مؤقت أيضًا؟

يفقد BullMQ المهام عندما يُسمح لـ Redis بحذف المفاتيح (eviction)، لأن كل مهمة وكل قفل وكل قائمة في الطابور هي مفتاح في Redis تعامله سياسة حذف مثل allkeys-lru كأنه ذاكرة مؤقتة يمكن الاستغناء عنها. المهمة المحذوفة من قائمة الانتظار لا تُعالَج أبدًا ولا تفشل أبدًا، فلا يصدر أي تنبيه. ويؤكد دليل الإنتاج في BullMQ أن noeviction هي قيمة maxmemory-policy الوحيدة التي تضمن سلوكًا صحيحًا للطوابير.

إعداد maxmemory-policy في Redis يسري على النسخة بالكامل: قواعد البيانات المنطقية التي تُختار بأمر SELECT تشترك فيه. الذاكرة المؤقتة تحتاج إلى الحذف والطابور يجب ألا يحذف أبدًا، لذا يشغّلهما فريقنا على نسختين منفصلتين من Redis. ومع noeviction يرفض Redis الممتلئ عمليات الكتابة بخطأ نفاد الذاكرة، فيتحول الفقدان الصامت للمهام إلى فشل واضح في queue.add()، وتصبح إعدادات removeOnComplete وremoveOnFail هي ميزانية الذاكرة الفعلية. ونفعّل أيضًا استمرارية AOF (appendonly yes) حتى لا تضيع المهام المنتظرة عند إعادة تشغيل Redis.

اتصالات العمال تحتاج إلى خيار ioredis التالي: maxRetriesPerRequest: null؛ يضبطه BullMQ للعمال افتراضيًا ويطبع تحذيرًا إذا تجاوزته نسخة ioredis مشتركة. أما اتصالات المنتِج فهي الحالة المعاكسة: يقترح توثيق BullMQ ضبط enableOfflineQueue: false على كائنات Queue، حتى يفشل طلب HTTP الذي يضيف مهمة بسرعة أثناء تعطل Redis بدلًا من أن يعلق.

لماذا نُفّذت مهمة BullMQ مرتين؟

تُنفَّذ مهمة BullMQ مرتين عندما تتوقف (stall): العامل الذي يعالجها لم يجدد قفلها في الوقت المحدد، فافترض BullMQ أن العامل توقف وسلّم المهمة لعامل آخر. كل مهمة نشطة تحمل قفلًا لمدة lockDuration (الافتراضي 30,000 ميلي ثانية)، ويجدده العامل كل lockRenewTime (الافتراضي 15,000 ميلي ثانية). التجديد يجري على حلقة أحداث Node.js، لذا فالمعالج الذي يحجب الحلقة — كإنشاء ملف PDF بشكل متزامن، أو تغيير حجم صورة، أو استدعاء JSON.parse() على بيانات ضخمة — يفوّت التجديد وهو لا يزال يعمل.

يعمل فاحص المهام المتوقفة كل stalledInterval (الافتراضي 30,000 ميلي ثانية) ويعيد المهمة إلى قائمة الانتظار؛ وإذا توقفت المهمة أكثر من maxStalledCount مرة (الافتراضي 1) تنتقل إلى مجموعة المهام الفاشلة بدلًا من ذلك. هذا التعافي صحيح عندما تُقتل الحاوية في منتصف المهمة، وخاطئ عندما يكون العامل الأصلي لا يزال حيًا، لأن نسختين من المهمة تعملان الآن في الوقت نفسه.

رفع قيمة lockDuration يخفي العَرَض ويؤخر التعافي من الأعطال الحقيقية بالمقدار نفسه، لذا ننقل العمل كثيف الاستهلاك للمعالج خارج الخيط الرئيسي. المعالج المعزول (sandboxed processor) هو ملف معالج يشغّله BullMQ في عملية فرعية منفصلة، أو في خيط عامل (worker thread) عند ضبط useWorkerThreads: true. في NestJS ننشئ كائن Worker عاديًا من BullMQ داخل مزوّد، ونمرر له مسار ملف المعالج المُترجَم، ونستدعي worker.close() في onModuleDestroy.

المقابل هو حقن التبعيات: المعالج المعزول يعمل خارج حاوية NestJS، لذا يُنشئ عملاء قاعدة البيانات والتخزين الخاصة به. نُبقي المعالجات المعزولة صغيرة، ونترك المهام المعتمدة على الإدخال والإخراج (I/O) على معالجات WorkerHost العادية.

كيف نمنع BullMQ من إضافة المهمة نفسها مرتين؟

يوفر BullMQ طريقتين مدمجتين لإسقاط المهام المكررة — معرّف jobId مخصص وخيار المهمة deduplication — وتختلفان في مدة استمرار فحص التكرار.

معرّف jobId مخصص: تُتجاهل المهمة المكررة ما دامت أي مهمة بالمعرّف نفسه مخزنة في Redis، بما في ذلك المهام المكتملة والفاشلة، ويُطلق BullMQ حدث duplicated. يناسب مهمة واحدة لكل كيان على الإطلاق.

خيار deduplication مع id: تُتجاهل المهمة المكررة ما دامت المهمة الأولى منتظرة أو مؤجلة أو نشطة، ويُطلق BullMQ حدث deduplicated. يناسب قاعدة "لا مهمة ثانية ما دامت الأولى قيد التنفيذ".

خيار deduplication مع id وttl: تُتجاهل المهمة المكررة ما دامت نافذة TTL مفتوحة. يناسب الحد من المعدل (throttling) وتجميع الدفعات المتتالية (debouncing).

فخ jobId المخصص هو مدة الاحتفاظ. عند ضبط removeOnComplete على عمر 86,400 ثانية، تبقى المهمة المكتملة ذات المعرّف receipt-42 في Redis يومًا كاملًا، وكل استدعاء add() بالمعرّف نفسه خلال ذلك اليوم يُتجاهل دون أي خطأ. كما لا يمكن أن تكون معرّفات المهام المخصصة نصوصًا من أرقام فقط مثل "123"؛ إذ يطلق BullMQ الخطأ "Custom Id cannot be integers".

أزال BullMQ 6 خيار debounce القديم. أصبح التجميع (debounce) الآن مهمة مؤجلة بقيمة ttl مطابقة مع ضبط extend وreplace على true، فتُنتج موجة من تعديلات المنتج إعادة فهرسة بحث واحدة بعد 5 ثوانٍ من آخر تعديل، وتحمل أحدث البيانات.

لا تجعل أيٌّ من الطريقتين المعالجة تحدث مرة واحدة بالضبط. فإعادة المحاولة والتوقف يشغّلان المهمة نفسها مرة أخرى، ولهذا تمرر معالجات البريد لدينا مفتاح idempotency مشتقًا من job.id إلى مزوّد البريد.

ما الذي يتعطل عند الترقية من BullMQ 5 إلى BullMQ 6؟

يزيل BullMQ 6.0.0 المهام المتكررة القديمة ويُطلق خطأً عند العثور على بياناتها الوصفية في Redis، لذا يجب نقل المهام المتكررة إلى Job Schedulers والتطبيق لا يزال يعمل على BullMQ 5. الـ Job Scheduler تعريف مُسمّى قابل للتحديث أو الإنشاء (upsert) يُنتج مهامًا وفق نمط cron عبر pattern أو فاصل زمني عبر every؛ واستدعاء upsertJobScheduler() مجددًا بالمعرّف نفسه يحدّث الجدول بدلًا من إنشاء جدول ثانٍ.

المهمة التي كانت تُضاف في BullMQ 5 بالخيار repeat: { pattern: '0 3 * * *', utc: true } تصبح queue.upsertJobScheduler('purge-expired-sessions', { pattern: '0 3 * * *', tz: 'UTC' }, { name: 'purge-expired-sessions' }). نستدعيها من onApplicationBootstrap() في خدمة NestJS؛ ولأن عملية upsert متساوية القوة، يمكن لكل نسخة من التطبيق تشغيلها عند الإقلاع.

الـ Job Schedulers موجودة بالفعل في BullMQ 5، لذا ترتيبنا هو: نشر كود الجدولة أولًا، ثم حذف التعريفات القديمة باستخدام queue.getRepeatableJobs() وqueue.removeRepeatableByKey()، وبعد ذلك فقط رفع الإصدار.

بقية التغييرات الجذرية في BullMQ 6 آلية: repeat.utc يصبح tz: 'UTC'؛ وخيار debounce وحدث debounced يحل محلهما deduplication وحدث deduplicated؛ وJob#discard() يحل محله UnrecoverableError؛ وWorker#resume() وQueue#resume() يجب انتظارهما بـ await؛ وioredis يجب تثبيتها صراحةً؛ وQueue#client وWorker#blockingClient يحل محلهما الـ backend الذي تعيده getBackend().

يقدّم BullMQ 6 أيضًا واجهات تخزين خلفية قابلة للتبديل (pluggable backends)، منها واجهة PostgreSQL اختيارية تشغّل واجهة Queue وWorker نفسها على PostgreSQL 13 أو أحدث عبر الدالة createPostgresBackend. ولا يزال توثيق BullMQ يصف Redis بأنه الخيار الافتراضي والأكثر اختبارًا في الإنتاج، ولم ننقل أي طابور إنتاجي إلى PostgreSQL حتى الآن.

الأسئلة الشائعة

س: هل يضمن BullMQ معالجة المهمة مرة واحدة بالضبط؟

ج: لا. يسلّم BullMQ المهام مرة واحدة على الأقل: إعادة المحاولة أو توقف المهمة أو تعطل العامل في منتصفها يشغّل المعالج مرة أخرى. اجعل كل معالج متساوي القوة، مثلًا بمفتاح idempotency مشتق من job.id.

س: هل يمكن أن يشترك BullMQ في نسخة Redis واحدة مع ذاكرة التخزين المؤقت للتطبيق؟

ج: ليس بأمان إذا كانت الذاكرة المؤقتة تعتمد على سياسة حذف مثل allkeys-lru. إعداد maxmemory-policy في Redis ينطبق على النسخة بالكامل، وBullMQ يحتاج إلى noeviction، لذا يجب أن يحصل الطابور على نسخة Redis خاصة به.

س: كم عدد المهام المنتهية التي يجب أن يحتفظ بها removeOnComplete؟

ج: احتفظ بعدد من المهام المكتملة يكفي لتتبع الأعمال الأخيرة، مقيدًا بالعمر والعدد معًا — نستخدم 24 ساعة و5,000 مهمة — واحتفظ بالمهام الفاشلة مدة أطول. ترك الخيارين دون ضبط يحتفظ بكل مهمة منتهية في Redis إلى الأبد.

س: متى نزيد قيمة lockDuration في BullMQ؟

ج: نادرًا. زيادة lockDuration تؤخر أيضًا التعافي من أعطال العمال بالمقدار نفسه، لذا انقل أولًا العمل كثيف الاستهلاك للمعالج إلى معالج معزول، وأبقِ حلقة الأحداث متفرغة لتجديد الأقفال.