SKIP LOCKED و Job Leasing: Postgres كـ job queue

import { pool, setup } from "./db.js";

await setup(20);

// ❌ كل worker بياخد أول job مش خلصانة
async function worker(name) {
  const done = [];
  for (;;) {
    const { rows } = await pool.query("SELECT id FROM jobs WHERE done_at IS NULL ORDER BY id LIMIT 1");
    if (!rows.length) return done;
    const id = rows[0].id;
    done.push(Number(id));
    await pool.query("UPDATE jobs SET done_at = now() WHERE id = $1", [id]);
  }
}

const [a, b, c] = await Promise.all(["A", "B", "C"].map(worker));
const all = [...a, ...b, ...c];
console.log("اتنفّذوا:", all.length, "| مكرر:", all.length - new Set(all).size);
await pool.end();

$ node main.js
اتنفّذوا: 46 | مكرر: 26

import { pool, setup, sleep } from "./db.js";

await setup(20);

// ✅ claim: خد job مش مقفولة، وحط عليها lease لـ 30 تانية
const CLAIM = `
  UPDATE jobs SET locked_until = now() + interval '30 seconds', attempts = attempts + 1
  WHERE id = (
    SELECT id FROM jobs
    WHERE done_at IS NULL AND run_at <= now() AND (locked_until IS NULL OR locked_until < now())
    ORDER BY run_at
    FOR UPDATE SKIP LOCKED
    LIMIT 1
  )
  RETURNING id, payload`;

async function worker(name) {
  const done = [];
  for (;;) {
    const { rows } = await pool.query(CLAIM);
    if (!rows.length) return done;
    await sleep(5); // الشغل نفسه
    await pool.query("UPDATE jobs SET done_at = now(), locked_until = NULL WHERE id = $1", [rows[0].id]);
    done.push(Number(rows[0].id));
  }
}

const results = await Promise.all(["A", "B", "C"].map(worker));
const all = results.flat();
console.log("اتنفّذوا:", all.length, "| مكرر:", all.length - new Set(all).size);
console.log("لكل worker:", results.map((r) => r.length).join(" / "));
await pool.end();

$ node main.js
اتنفّذوا: 20 | مكرر: 0
لكل worker: 7 / 7 / 6

import { pool, setup, sleep } from "./db.js";

await setup(1);
const claim = (lease) => pool.query(`
  UPDATE jobs SET locked_until = now() + $1::interval, attempts = attempts + 1
  WHERE id = (SELECT id FROM jobs WHERE done_at IS NULL AND (locked_until IS NULL OR locked_until < now())
              ORDER BY run_at FOR UPDATE SKIP LOCKED LIMIT 1)
  RETURNING id, attempts`, [lease]);

// worker 1 أخد الـ job ومات (من غير ما يعلّم done)
console.log("worker1:", (await claim("300 milliseconds")).rows);
console.log("worker2 فورًا:", (await claim("300 milliseconds")).rows, "← مقفولة بالـ lease");
await sleep(400);
console.log("worker2 بعد ما الـ lease خلص:", (await claim("300 milliseconds")).rows);
await pool.end();

$ node main.js
worker1: [ { id: '1', attempts: 1 } ]
worker2 فورًا: [] ← مقفولة بالـ lease
worker2 بعد ما الـ lease خلص: [ { id: '1', attempts: 2 } ]

import { pool, setup, sleep } from "./db.js";

await setup(6);
// jobs 3 و 5 بايظين: بيفشلوا كل مرة
const handlers = {
  email: async (p) => {
    await sleep(5);
    if (p.n === 3 || p.n === 5) throw new Error(`smtp refused #${p.n}`);
  },
};
const MAX_ATTEMPTS = 3;


await pool.query("ALTER TABLE jobs ADD COLUMN IF NOT EXISTS dead_at timestamptz");

const CLAIM = `
  UPDATE jobs SET locked_until = now() + interval '30 seconds', attempts = attempts + 1
  WHERE id = (
    SELECT id FROM jobs
    WHERE done_at IS NULL AND dead_at IS NULL AND run_at <= now()
      AND (locked_until IS NULL OR locked_until < now())
    ORDER BY run_at FOR UPDATE SKIP LOCKED LIMIT 1
  )
  RETURNING id, kind, payload, attempts`;

async function worker() {
  for (;;) {
    const { rows } = await pool.query(CLAIM);
    if (!rows.length) {
      const waiting = await pool.query("SELECT 1 FROM jobs WHERE done_at IS NULL AND dead_at IS NULL LIMIT 1");
      if (!waiting.rows.length) return;
      await sleep(30);
      continue;
    }
    const job = rows[0];
    try {
      await handlers[job.kind](job.payload);
      await pool.query("UPDATE jobs SET done_at = now(), locked_until = NULL WHERE id = $1", [job.id]);
    } catch (err) {
      if (job.attempts >= MAX_ATTEMPTS) {
        await pool.query("UPDATE jobs SET dead_at = now(), last_error = $2 WHERE id = $1", [job.id, err.message]);
      } else {
        await pool.query(
          "UPDATE jobs SET run_at = now() + $2 * interval '50 milliseconds', locked_until = NULL, last_error = $3 WHERE id = $1",
          [job.id, job.attempts, err.message],
        );
      }
    }
  }
}

await Promise.all([worker(), worker()]);
const { rows } = await pool.query(
  "SELECT id, attempts, done_at IS NOT NULL AS done, dead_at IS NOT NULL AS dead, last_error FROM jobs ORDER BY id",
);
console.table(rows.map((r) => ({ ...r, id: Number(r.id) })));
await pool.end();

$ node main.js
┌─────────┬────┬──────────┬───────┬───────┬───────────────────┐
│ (index) │ id │ attempts │ done  │ dead  │ last_error        │
├─────────┼────┼──────────┼───────┼───────┼───────────────────┤
│ 0       │ 1  │ 1        │ true  │ false │ null              │
│ 1       │ 2  │ 1        │ true  │ false │ null              │
│ 2       │ 3  │ 3        │ false │ true  │ 'smtp refused #3' │
│ 3       │ 4  │ 1        │ true  │ false │ null              │
│ 4       │ 5  │ 3        │ false │ true  │ 'smtp refused #5' │
│ 5       │ 6  │ 1        │ true  │ false │ null              │
└─────────┴────┴──────────┴───────┴───────┴───────────────────┘