ابنِ queue في ٦٠ دقيقة

الـ queue الأدنى القابل للتشغيل#

queue المزامنة، في جوهره، آلة حالة صغيرة على قائمة من الوظائف. كلّ وظيفة تتحرّك عبر pending ← in_progress ← done أو failed، ومهمّة الـ queue هي ضمان أنّ هذه الحركة آمنة عند إعادة المحاولة: إن سقطت الشبكة في منتصف flush، يجب ألّا تفقد عملاً، ويجب ألّا تكتب مرّتَين.

سنبدأ بأصغر نوع يلتقط هذا:

type JobStatus = 'pending' | 'in_progress' | 'done' | 'failed';

interface Job<T = unknown> {
  id: string;
  idempotencyKey: string;
  payload: T;
  status: JobStatus;
  attempts: number;
  lastError?: string;
}

حقلان يكسبان مكانهما هنا: idempotencyKey و attempts. الأوّل عقد مع الخادم (نفس المفتاح عند إعادة المحاولة يجب ألّا ينتج كتابة ثانية). الثاني عقد مع نفسك (تحتاج شرط توقّف).

الخطوة ١: queue في الذاكرة#

النسخة الأولى تخزّن الوظائف في Map<string, Job>. هذا يكفي لاختبار آلة الحالة وحلقة إعادة المحاولة دون إدخال التخزين في الصورة. التبنّي الفعليّ يستبدل هذا بـ SQLite، IndexedDB، أو ما توفّره المنصّة، لكن واجهة الـ queue لا يجب أن تهتمّ.

class Queue<T> {
  private jobs = new Map<string, Job<T>>();

  enqueue(payload: T, idempotencyKey: string): Job<T> {
    const id = crypto.randomUUID();
    const job: Job<T> = {
      id,
      idempotencyKey,
      payload,
      status: 'pending',
      attempts: 0,
    };
    this.jobs.set(id, job);
    return job;
  }

  pending(): Job<T>[] {
    return [...this.jobs.values()].filter((j) => j.status === 'pending');
  }
}

شيئان للملاحظة. أوّلاً، الـ queue نفسه لا يفعل I/O. حلقة الـ flush مسؤوليّة منفصلة. ثانياً، enqueue متزامن. enqueue غير متزامن فخّ: يعني أنّ المُتصِل لا يستطيع التفكير ذرّيّاً في "هل دخل هذا أم لا".

الخطوة ٢: حلقة الـ flush#

حلقة الـ flush تمشي على الوظائف pending، تعطي كلّ واحدة لدالّة pusher تعيد promise، وتُحدّث الحالة من النتيجة. العقد مع الـ pusher صغير: إمّا يحلّ، أو يرفض بخطأ.

type Pusher<T> = (job: Job<T>) => Promise<void>;

async flush(pusher: Pusher<T>): Promise<void> {
  for (const job of this.pending()) {
    job.status = 'in_progress';
    job.attempts += 1;
    try {
      await pusher(job);
      job.status = 'done';
    } catch (e) {
      job.status = job.attempts >= 5 ? 'failed' : 'pending';
      job.lastError = String(e);
    }
  }
}

الفرع في كتلة الـ catch هو المكان الوحيد الذي تعيش فيه سياسة الـ back-off. خمس محاولات ونُعلِّم الوظيفة failed ونمشي. التطبيق الحقيقيّ سيريد أيضاً تأخيراً أُسّيّاً بين المحاولات؛ أبسط نسخة هي setTimeout(flush, 2 ** attempts * 1000) من المُتصِل، ليس من داخل الـ queue.

الخطوة ٣: حزام الاختبار#

دون اختبار، لا تستطيع إثبات أنّ هذا صحيح عند إعادة المحاولة. حزام الاختبار pusher مزيّف يحاكي إخفاقات الشبكة بالـ index. ضع الملفّ في src/queue.test.ts:

import { describe, it, expect, vi } from 'vitest';
import { Queue } from './queue';

describe('Queue', () => {
  it('retries failed jobs until they succeed', async () => {
    const q = new Queue<string>();
    q.enqueue('A', 'key-A');

    let failures = 2;
    const pusher = vi.fn(async () => {
      if (failures-- > 0) throw new Error('network');
    });

    await q.flush(pusher);
    await q.flush(pusher);
    await q.flush(pusher);

    expect(pusher).toHaveBeenCalledTimes(3);
    expect(q.pending()).toHaveLength(0);
  });

  it('marks the job failed after 5 attempts', async () => {
    const q = new Queue<string>();
    q.enqueue('B', 'key-B');
    const pusher = vi.fn(async () => { throw new Error('always'); });

    for (let i = 0; i < 5; i++) await q.flush(pusher);

    expect(pusher).toHaveBeenCalledTimes(5);
    expect(q.pending()).toHaveLength(0);
  });
});

شغّله بـ pnpm vitest run. كلا الاختبارَين يجب أن ينجحا.

الخطوة ٤: عقد الـ idempotency#

الآن، الـ idempotencyKey مخزَّن لكن غير مستخدَم. التغيير التالي يجعله يكسب مكانه: الـ pusher يستخدم المفتاح كـ HTTP header، والخادم يستخدمه لإلغاء تكرار الكتابات. أدناه النسخة الإنتاجيّة من pusher.

async function pushToServer(job: Job<{ url: string; body: unknown }>) {
  const res = await fetch(job.payload.url, {
    method: 'POST',
    headers: {
      'content-type': 'application/json',
      'idempotency-key': job.idempotencyKey,
    },
    body: JSON.stringify(job.payload.body),
  });
  if (!res.ok) throw new Error(`HTTP ${res.status}`);
}

عقد الخادم هو: نفس المفتاح، نفس الـ payload، المرّة الثانية ← نفس الاستجابة، لا كتابة ثانية. هذا ما يجعل "أعِد المحاولة حتّى تعمل" آمناً. بدونه، كلّ إعادة محاولة رمي زهر على تكرار.

الخطوة ٥: أين يفشل هذا#

هذا الـ queue يفعل الأشياء الصحيحة لأحمال صغيرة. هو أيضاً يكذب حول ثلاثة أشياء ستهتمّ بها في النهاية:

1. الديمومة. أعد تحميل الصفحة والـ queue ذهب. التطبيقات الحقيقيّة تحتاج طبقة تخزين تنجو من إعادة التحميل.
2. التوازي. حلقة الـ flush تشغّل الوظائف بالتسلسل. التطبيقات الحقيقيّة تريد توازياً محدوداً، ليس لانهائيّاً، ليس واحداً-في-المرّة.
3. حلّ التعارض. حين يضع عميلان كتابات متناقضة في الـ queue، الخادم عليه أن يختار. الـ queue لا يستطيع.

الثلاثة بالضبط هي ما تغطّيه دورة محرّكات المزامنة من طرف لطرف الكاملة، بنفس أسلوب الكود ونفس شكل اتّخاذ القرار. هذا المختبر الفصل الأوّل؛ الدورة الباقي.

إغلاق الحلقة#

يجب أن يكون لديك الآن ملفّ queue، ملفّ اختبار، و pusher واعٍ بالـ idempotency. شغّل الاختبارات مرّة أخرى، ثمّ وجّه الـ queue إلى نقطة نهاية حقيقيّة من اختيارك. الساعة انتهت. الـ queue لك.