# tasks.py, Define the task (consumer side)
from celery import Celery
import resend

# Connect Celery to Redis broker
app = Celery('tasks', broker='redis://localhost:6379/0')

# Decorate the function as a Celery task
@app.task(bind=True, max_retries=5, default_retry_delay=60)
def send_verification_email(self, user_id: str, email: str, token: str):
    """
    Called by the worker process.
    self.retry() implements exponential backoff automatically.
    """
    try:
        params = {
            "from": "noreply@myapp.com",
            "to": [email],
            "subject": "Verify your email",
            "html": build_email_template(token),
        }
        resend.Emails.send(params)
    except Exception as exc:
        # Exponential backoff: 60s, 120s, 240s, 480s, 960s
        countdown = 60 * (2 ** self.request.retries)
        raise self.retry(exc=exc, countdown=countdown)

// worker.js, Consumer, BullMQ calls this for each job
import { Worker } from 'bullmq'
import { Resend } from 'resend'

const connection = { host: 'localhost', port: 6379 }
const resend = new Resend(process.env.RESEND_API_KEY)

// The queue NAME is the contract between producer and consumer; both
// sides must spell it the same way. That is BullMQ's version of Asynq's
// task type constant.
export const worker = new Worker(
  'email',
  async (job) => {
    // 1. The payload is already a JS object; BullMQ serialized it for us
    const { userId, email, token } = job.data

    // 2. Execute: call email provider API
    console.log('sending verification email', userId, email)
    await resend.emails.send({
      from: 'noreply@myapp.com',
      to: [email],
      subject: 'Verify your email',
      html: buildEmailTemplate(token),
    })

    // 3. Returning normally ACKs the job (task done). THROWING is what
    //    triggers a retry, so never swallow an error in here: a caught
    //    exception looks exactly like success to the queue.
  },
  { connection, concurrency: 10 },
)

worker.on('failed', (job, err) => {
  console.error(`job ${job?.id} failed (attempt ${job?.attemptsMade})`, err)
})

import { Worker, UnrecoverableError, type Job } from 'bullmq'

// One exported type, imported by BOTH the producer and this worker. It is
// the cheapest fix for the single most common task-queue bug: a producer
// that starts sending a renamed field while the worker still reads the
// old one. Here that mismatch is a compile error, not a 3am page.
export interface EmailJob {
  userId: string
  email: string
  token: string
}

export const worker = new Worker<EmailJob>(
  'email',
  async (job: Job<EmailJob>) => {
    const { userId, email, token } = job.data

    if (typeof email !== 'string') {
      // Non-retryable: no amount of retrying fixes a malformed payload.
      // UnrecoverableError is BullMQ's asynq.SkipRetry.
      throw new UnrecoverableError('malformed payload: email missing')
    }

    console.log('sending verification email', userId, email)
    await resend.emails.send({
      from: 'noreply@myapp.com',
      to: [email],
      subject: 'Verify your email',
      html: buildEmailTemplate(token),
    })
    // Returning normally ACKs; throwing anything else triggers a retry.
  },
  { connection, concurrency: 10 },
)

// Java has no single dominant task queue, so this is the mechanism itself:
// a Redis Stream with a consumer group, which is exactly what Asynq builds
// on. Section 6 of this chapter describes the XADD/XREADGROUP/XACK cycle
// you are about to see in code.
public record EmailPayload(String userId, String email, String token) {}

@Component
public class EmailTaskHandler
        implements StreamListener<String, MapRecord<String, String, String>> {

    static final String STREAM = "tasks:email";
    static final String GROUP = "email-workers";
    static final String DLQ = "tasks:email:dead";

    private final StringRedisTemplate redis;
    private final ObjectMapper json;
    private final EmailService emailSvc; // injected dependency

    EmailTaskHandler(StringRedisTemplate redis, ObjectMapper json, EmailService emailSvc) {
        this.redis = redis;
        this.json = json;
        this.emailSvc = emailSvc;
    }

    @Override
    public void onMessage(MapRecord<String, String, String> record) {
        EmailPayload p;

        // 1. Deserialize JSON -> Java record
        try {
            p = json.readValue(record.getValue().get("payload"), EmailPayload.class);
        } catch (JsonProcessingException e) {
            // Non-retryable: this payload will never parse. ACK it so it
            // stops being redelivered, and park it in the dead letter stream.
            redis.opsForStream().add(DLQ, record.getValue());
            redis.opsForStream().acknowledge(STREAM, GROUP, record.getId());
            return;
        }

        // 2. Execute: call email provider API
        try {
            log.info("sending verification email user_id={} email={}", p.userId(), p.email());
            emailSvc.sendVerification(p.email(), p.token());
        } catch (RuntimeException e) {
            // Retryable: deliberately do NOT acknowledge. The message stays
            // in the group's pending list and another worker reclaims it
            // via XAUTOCLAIM once the idle timeout passes.
            log.warn("send failed; leaving unacked for redelivery", e);
            return;
        }

        // 3. XACK -> the task is done and leaves the pending list
        redis.opsForStream().acknowledge(STREAM, GROUP, record.getId());
    }
}

# signup.py, Producer side (inside your API handler)
from fastapi import FastAPI
from tasks import send_verification_email

app = FastAPI()

@app.post("/signup")
async def signup(body: SignupRequest):
    # 1. Validate, hash password, save user to DB
    user = await create_user(body)
    token = generate_verification_token(user.id)

    # 2. Enqueue, returns instantly, does NOT wait for email
    send_verification_email.delay(
        user_id=user.id,
        email=user.email,
        token=token
    )

    # 3. Return 201 immediately
    return {"message": "Account created. Check your email."}, 201

// signup.js, Producer side (inside your API handler)
import { Queue } from 'bullmq'

// Same queue name the worker listens on
const emailQueue = new Queue('email', { connection: { host: 'localhost', port: 6379 } })

app.post('/signup', async (req, res) => {
  // 1. Validate, hash password, save user to DB
  const user = await createUser(req.body)
  const token = generateVerificationToken(user.id)

  // 2. Enqueue, returns instantly, does NOT wait for email
  await emailQueue.add(
    'send_verification',
    { userId: user.id, email: user.email, token },
    { attempts: 5, backoff: { type: 'exponential', delay: 60_000 } },
  )

  // 3. Return 201 immediately
  res.status(201).json({ message: 'Account created. Check your email.' })
})

import { Queue } from 'bullmq'
import type { Request, Response } from 'express'
import type { EmailJob } from './worker'

// The SAME EmailJob the worker imports. `add` now refuses a payload the
// worker cannot read, which is the entire point of importing it here.
const emailQueue = new Queue<EmailJob>('email', {
  connection: { host: 'localhost', port: 6379 },
})

export async function signup(req: Request, res: Response): Promise<void> {
  // 1. Validate, hash password, save user to DB
  const user = await createUser(req.body)
  const token = generateVerificationToken(user.id)

  // 2. Enqueue, returns instantly, does NOT wait for email
  await emailQueue.add(
    'send_verification',
    { userId: user.id, email: user.email, token },
    {
      attempts: 5,
      backoff: { type: 'exponential', delay: 60_000 }, // 60s, 120s, 240s...
      removeOnComplete: 1_000, // keep the last 1k, not every job forever
    },
  )

  // 3. Return 201 immediately
  res.status(201).json({ message: 'Account created. Check your email.' })
}

@Service
class EmailTaskProducer {
    private final StringRedisTemplate redis;
    private final ObjectMapper json;

    EmailTaskProducer(StringRedisTemplate redis, ObjectMapper json) {
        this.redis = redis;
        this.json = json;
    }

    String enqueue(EmailPayload payload) throws JsonProcessingException {
        // XADD tasks:email * payload <json>, returns immediately
        RecordId id = redis.opsForStream().add(StreamRecords.newRecord()
            .in(EmailTaskHandler.STREAM)
            .ofMap(Map.of("payload", json.writeValueAsString(payload))));
        return id.getValue(); // the task id, usable for monitoring
    }
}

@RestController
class SignupController {

    @PostMapping("/signup")
    @ResponseStatus(HttpStatus.CREATED) // 3. Return 201 immediately
    Map<String, String> signup(@RequestBody SignupRequest body) throws JsonProcessingException {
        // 1. Validate, hash password, save user to DB
        User user = createUser(body);
        String token = generateVerificationToken(user.id());

        // 2. Enqueue, returns instantly, does NOT wait for email
        producer.enqueue(new EmailPayload(user.id(), user.email(), token));

        return Map.of("message", "Account created. Check your email.");
    }
}

# Celery's worker is a command, not code you write. The same tasks.py is
# imported by both your API (to enqueue) and this process (to execute).
#
#   celery -A tasks worker \
#       --concurrency=10 \
#       --queues=email,default,low \
#       --loglevel=info
#
# --concurrency defaults to the CPU count, which is the wrong default for
# IO-bound work like sending email. See chapter 20.

# Priority is expressed by ROUTING tasks to named queues:
app.conf.task_routes = {
    "tasks.send_verification_email": {"queue": "email"},
    "tasks.generate_monthly_report": {"queue": "low"},
}

# Then run one worker per priority class, and give the important queue
# more processes rather than trusting a single worker to interleave fairly:
#   celery -A tasks worker --queues=email   --concurrency=6
#   celery -A tasks worker --queues=default --concurrency=3
#   celery -A tasks worker --queues=low     --concurrency=1

// This file is its own process: `node worker.js`, never imported by the API.
import { Worker } from 'bullmq'

const connection = { host: 'localhost', port: 6379 }

// concurrency is how many jobs this ONE process runs at a time. Node is
// single-threaded, so this is only a win for IO-bound work; for CPU-bound
// tasks use several processes (or a Sandboxed Processor). See chapter 20.
new Worker('email', emailProcessor, { connection, concurrency: 10 })
new Worker('default', defaultProcessor, { connection, concurrency: 3 })
new Worker('low', lowProcessor, { connection, concurrency: 1 })

// Graceful shutdown: stop accepting new jobs, let in-flight ones finish.
// Without this, a deploy kills jobs mid-send and they are redelivered,
// which is exactly why handlers must be idempotent. See chapter 16.
process.on('SIGTERM', async () => {
  await Promise.all([emailWorker.close(), defaultWorker.close(), lowWorker.close()])
  process.exit(0)
})

import { Worker } from 'bullmq'
import type { EmailJob } from './types'

const connection = { host: 'localhost', port: 6379 }

// BullMQ has no cross-queue priority setting: a Worker drains exactly one
// queue. Priority is therefore a RESOURCE decision, you give the queue
// that matters more concurrency, and run it in more replicas.
const workers = [
  new Worker<EmailJob>('email', emailProcessor, { connection, concurrency: 10 }),
  new Worker('default', defaultProcessor, { connection, concurrency: 3 }),
  new Worker('low', lowProcessor, { connection, concurrency: 1 }),
]

process.on('SIGTERM', async () => {
  // close() waits for in-flight jobs; a hard kill would let them be
  // redelivered, which only a genuinely idempotent handler survives.
  await Promise.all(workers.map((w) => w.close()))
  process.exit(0)
})

@Configuration
public class WorkerConfig {

    @Bean
    StreamMessageListenerContainer<String, MapRecord<String, String, String>> emailWorker(
            RedisConnectionFactory cf, EmailTaskHandler handler) {

        var options = StreamMessageListenerContainer
            .StreamMessageListenerContainerOptions.builder()
            .pollTimeout(Duration.ofSeconds(2))
            // The thread pool IS the concurrency knob: 10 tasks at a time.
            .executor(Executors.newFixedThreadPool(10))
            .build();

        var container = StreamMessageListenerContainer.create(cf, options);

        // Reading as a CONSUMER GROUP is what makes this a work queue
        // rather than a broadcast: each message goes to exactly ONE
        // consumer, and anything unacked can be reclaimed by another.
        container.receive(
            Consumer.from(EmailTaskHandler.GROUP, "worker-" + UUID.randomUUID()),
            StreamOffset.create(EmailTaskHandler.STREAM, ReadOffset.lastConsumed()),
            handler);

        container.start();
        return container;
    }
}

// Priority: one container per queue, sized by importance, exactly as the
// Go version weights its queues 6 / 3 / 1.
//
// Higher-level options exist if you would rather not hold the stream
// yourself: JobRunr (annotation-driven, SQL/Mongo-backed), Spring Batch
// for chunked bulk jobs, or Spring Cloud AWS for @SqsListener.

# celery_config.py, Celery Beat schedule
from celery.schedules import crontab

CELERYBEAT_SCHEDULE = {
    # Send weekly report every Sunday midnight
    'weekly-report': {
        'task': 'tasks.send_weekly_report',
        'schedule': crontab(hour=0, minute=0, day_of_week='sunday'),
    },
    # Cleanup orphan sessions on the 1st of each month
    'session-cleanup': {
        'task': 'tasks.cleanup_orphan_sessions',
        'schedule': crontab(hour=3, minute=0, day_of_month='1'),
    },
}

# Beat is its own process, separate from the workers:
#   celery -A tasks beat --loglevel=info

// BullMQ has no separate scheduler process: a repeatable job is just a
// job with a cron pattern, and Redis deduplicates it by repeat key. That
// makes running this file on every replica harmless.
import { Queue } from 'bullmq'

const reports = new Queue('reports', { connection })

// Send weekly report every Sunday at midnight
await reports.add('weekly', null, {
  repeat: { pattern: '0 0 * * 0', tz: 'UTC' },
  jobId: 'report:weekly', // stable id, re-adding it does not duplicate
})

// Cleanup orphan sessions every 1st of the month
await reports.add('sessionCleanup', null, {
  repeat: { pattern: '0 3 1 * *', tz: 'UTC' },
  jobId: 'sessions:cleanup',
})

// Always set `tz`. A schedule that silently follows the server's local
// time will shift by an hour twice a year, and the bug surfaces months
// later as "the Sunday report went out on Saturday night".

import { Queue } from 'bullmq'

interface Schedule {
  name: string
  pattern: string
  jobId: string
}

// Listing the schedules as DATA rather than as a sequence of calls is what
// makes them reviewable: one array to read in a PR, and one loop that
// cannot forget the tz or the stable jobId on the entry someone adds next.
const SCHEDULES: Schedule[] = [
  { name: 'weekly', pattern: '0 0 * * 0', jobId: 'report:weekly' },
  { name: 'sessionCleanup', pattern: '0 3 1 * *', jobId: 'sessions:cleanup' },
]

const reports = new Queue('reports', { connection })

for (const { name, pattern, jobId } of SCHEDULES) {
  await reports.add(name, null, { repeat: { pattern, tz: 'UTC' }, jobId })
}

@Configuration
@EnableScheduling
public class SchedulerConfig {

    private final EmailTaskProducer producer;

    SchedulerConfig(EmailTaskProducer producer) {
        this.producer = producer;
    }

    // Send weekly report every Sunday at midnight.
    // Note the six fields: Spring's cron starts with SECONDS, so the
    // five-field Unix expression "0 0 * * 0" becomes "0 0 0 * * SUN".
    @Scheduled(cron = "0 0 0 * * SUN", zone = "UTC")
    void enqueueWeeklyReport() {
        // The scheduler only ENQUEUES; an ordinary worker runs it.
        producer.enqueue("report:weekly", null);
    }

    // Cleanup orphan sessions every 1st of the month
    @Scheduled(cron = "0 0 3 1 * *", zone = "UTC")
    void enqueueSessionCleanup() {
        producer.enqueue("sessions:cleanup", null);
    }
}

// @Scheduled fires on EVERY instance, so with two replicas every task is
// enqueued twice. In a multi-replica deployment take a lock first, with
// ShedLock (@SchedulerLock) or Quartz in clustered mode.

# Python, Idempotency key pattern with Redis
import redis
import hashlib

r = redis.Redis(host='localhost', port=6379)

@app.task(bind=True, max_retries=5)
def send_verification_email(self, user_id: str, email: str, token: str):
    # Build idempotency key from task inputs
    key = f"idem:verify_email:{user_id}:{token}"

    # SET NX = set only if Not eXists; EX = expire after 1 hour
    # Returns True if we are the FIRST execution, False if already done
    acquired = r.set(key, "done", nx=True, ex=3600)
    if not acquired:
        # Already processed, skip silently (idempotent return)
        return {"status": "already_sent"}

    # First time, actually send the email
    try:
        _send_email_via_provider(email, token)
    except Exception as exc:
        # Delete the key so next retry can attempt again
        r.delete(key)
        raise self.retry(exc=exc, countdown=60 * 2 ** self.request.retries)

// Node, Idempotency key pattern with Redis
new Worker('email', async (job) => {
  const { userId, email, token } = job.data

  // Build idempotency key from task inputs
  const key = `idem:verify_email:${userId}:${token}`

  // NX = set only if Not eXists; EX = expire after 1 hour.
  // set() returns 'OK' on the FIRST execution, and null if already done.
  const acquired = await redis.set(key, 'done', { NX: true, EX: 3600 })
  if (acquired === null) {
    return { status: 'already_sent' } // skip silently (idempotent return)
  }

  // First time, actually send the email
  try {
    await sendEmailViaProvider(email, token)
  } catch (err) {
    await redis.del(key) // delete the key so the next retry can attempt again
    throw err // throwing is what schedules the retry
  }
}, { connection })

// Worth naming the gap in this shape before you rely on it: if the process
// dies AFTER the SET NX but BEFORE the send lands, the key is held and the
// retry skips the email entirely. The one-hour TTL is what bounds the
// damage. The stronger fix is the one chapter 07 uses for payments, store
// the RESULT under the key instead of a bare "done" marker, so a duplicate
// can return the original outcome rather than guessing.
async function handler(job: Job<EmailJob>): Promise<{ status: string }> {
  const { userId, email, token } = job.data
  const key = `idem:verify_email:${userId}:${token}`

  const acquired: string | null = await redis.set(key, 'pending', { NX: true, EX: 3600 })
  if (acquired === null) {
    return { status: 'already_sent' } // skip silently (idempotent return)
  }

  try {
    const messageId = await sendEmailViaProvider(email, token)
    // Overwrite the claim with the real result, still under the same TTL.
    await redis.set(key, JSON.stringify({ messageId }), { EX: 3600 })
    return { status: 'sent' }
  } catch (err) {
    await redis.del(key) // let the next retry attempt again
    throw err
  }
}

// Java, Idempotency key pattern with Redis
@Override
public void onMessage(MapRecord<String, String, String> record) {
    EmailPayload p = parse(record); // as in the consumer example above

    // Build idempotency key from task inputs
    String key = "idem:verify_email:%s:%s".formatted(p.userId(), p.token());

    // setIfAbsent IS SET NX, and the Duration is the EX. It returns TRUE
    // only for the FIRST execution, FALSE if an earlier attempt got there.
    Boolean acquired = redis.opsForValue()
        .setIfAbsent(key, "done", Duration.ofHours(1));

    if (!Boolean.TRUE.equals(acquired)) {
        // Already processed, ACK cleanly (idempotent return). Note the
        // Boolean.TRUE.equals: `acquired` is a nullable Boolean, and
        // unboxing a null here would throw instead of skipping.
        redis.opsForStream().acknowledge(STREAM, GROUP, record.getId());
        return;
    }

    // First time, actually send the email
    try {
        emailSvc.sendVerification(p.email(), p.token());
        redis.opsForStream().acknowledge(STREAM, GROUP, record.getId());
    } catch (RuntimeException e) {
        redis.delete(key); // release the key so the redelivery can try again
        log.warn("send failed; leaving unacked for redelivery", e);
    }
}

# Python, Idempotent multi-step task with full transaction rollback
@app.task(bind=True, max_retries=5)
def delete_account(self, user_id: str):
    # Check if already deleted (idempotency guard)
    if not db.user_exists(user_id):
        return  # already deleted on a previous attempt, ACK cleanly

    try:
        # session.begin() COMMITs on a clean exit and ROLLBACKs on any
        # exception, so the whole multi-step delete is one unit.
        with Session(engine) as session, session.begin():
            delete_user_projects(session, user_id)
            delete_user_sessions(session, user_id)
            delete_user_assets(session, user_id)
            delete_user_account(session, user_id)
    except SQLAlchemyError as exc:
        # Nothing was written, so the retry starts from a clean database
        raise self.retry(exc=exc, countdown=60 * 2 ** self.request.retries)

// Node, Idempotent multi-step task with full transaction rollback
new Worker('account', async (job) => {
  const { userId } = job.data

  // Check if already deleted (idempotency guard)
  if (!(await db.userExists(userId))) {
    return // already deleted on a previous attempt, ACK cleanly
  }

  // Wrap all DB writes in a single transaction, on ONE connection
  const tx = await pool.connect()
  try {
    await tx.query('BEGIN')
    await deleteUserProjects(tx, userId)
    await deleteUserSessions(tx, userId)
    await deleteUserAssets(tx, userId)
    await deleteUserAccount(tx, userId)
    await tx.query('COMMIT') // all steps succeeded -> ACK
  } catch (err) {
    await tx.query('ROLLBACK') // full rollback; task retried from scratch
    throw err
  } finally {
    tx.release()
  }
}, { connection })

// The type on this array is doing real work: every step takes a `tx`, so
// a function that does NOT take one cannot be added to the list. That
// matters more than it looks, because a rollback only undoes DATABASE
// writes. Slip an email send or an S3 delete in here and the retry does
// it a second time with nothing to roll back.
const steps: Array<(tx: PoolClient, userId: string) => Promise<void>> = [
  deleteUserProjects,
  deleteUserSessions,
  deleteUserAssets,
  deleteUserAccount,
]

async function handleDeleteAccount(job: Job<{ userId: string }>): Promise<void> {
  const { userId } = job.data

  // Check if already deleted (idempotency guard)
  if (!(await db.userExists(userId))) {
    return // already deleted on a previous attempt, ACK cleanly
  }

  // withTransaction (chapter 08) commits on return, rolls back on throw
  await withTransaction(pool, async (tx) => {
    for (const step of steps) {
      await step(tx, userId) // any throw -> full rollback -> retry from scratch
    }
  })
}

// Java, Idempotent multi-step task with full transaction rollback
@Service
public class AccountTaskHandler {

    // @Transactional wraps the WHOLE method: every write below commits
    // together, or a thrown exception rolls all of them back and the
    // redelivered task starts against a clean database.
    @Transactional
    public void handleDeleteAccount(DeleteAccountPayload p) {
        // Check if already deleted (idempotency guard)
        if (!db.userExists(p.userId())) {
            return; // already deleted on a previous attempt, ACK cleanly
        }

        db.deleteUserProjects(p.userId());
        db.deleteUserSessions(p.userId());
        db.deleteUserAssets(p.userId());
        db.deleteUserAccount(p.userId());
        // returning commits; any RuntimeException triggers a full rollback
    }
}

// Both chapter 08 traps are genuinely dangerous here, not academic:
//  1. The proxy. Calling handleDeleteAccount() from another method of
//     THIS class runs it with NO transaction at all, so a failure halfway
//     down leaves an account that is half deleted and cannot be retried.
//  2. Checked exceptions do not roll back by default, which commits the
//     partial delete. Be explicit:
//     @Transactional(rollbackFor = Exception.class)

# Python, SQS FIFO with deduplication
import boto3

sqs = boto3.client("sqs", region_name="us-east-1")

sqs.send_message(
    QueueUrl="https://sqs.us-east-1.amazonaws.com/123/orders.fifo",
    MessageBody=payload,
    MessageGroupId="order-processing",   # ordering group
    MessageDeduplicationId=order_id,     # unique per business event
)
# Sending the same order_id twice within 5 min -> second is silently dropped

// Node, SQS FIFO with deduplication
import { SQSClient, SendMessageCommand } from '@aws-sdk/client-sqs'

const sqs = new SQSClient({ region: 'us-east-1' })

await sqs.send(new SendMessageCommand({
  QueueUrl: 'https://sqs.us-east-1.amazonaws.com/123/orders.fifo',
  MessageBody: payload,
  MessageGroupId: 'order-processing', // ordering group
  MessageDeduplicationId: orderId,    // unique per business event
}))
// Sending the same orderId twice within 5 min -> second is silently dropped

import { SQSClient, SendMessageCommand } from '@aws-sdk/client-sqs'

const sqs = new SQSClient({ region: 'us-east-1' })

// The dedup id has to come from the BUSINESS event, never from the
// attempt: a randomUUID() per send makes every retry look like a new
// message and deduplicates nothing. Typing it as the order id and not as
// a bare `string` is a cheap way to keep that honest.
export async function enqueueOrder(orderId: OrderId, payload: string): Promise<void> {
  await sqs.send(new SendMessageCommand({
    QueueUrl: process.env.ORDERS_QUEUE_URL,
    MessageBody: payload,
    MessageGroupId: 'order-processing', // ordering group
    MessageDeduplicationId: orderId,    // unique per business event
  }))
}
// Sending the same orderId twice within 5 min -> second is silently dropped

// Java, SQS FIFO with deduplication
SqsClient sqs = SqsClient.builder().region(Region.US_EAST_1).build();

sqs.sendMessage(SendMessageRequest.builder()
    .queueUrl("https://sqs.us-east-1.amazonaws.com/123/orders.fifo")
    .messageBody(payload)
    .messageGroupId("order-processing") // ordering group
    .messageDeduplicationId(orderId)    // unique per business event
    .build());
// Sending the same orderId twice within 5 min -> second is silently dropped

# rate_limiter.py, Redis token bucket implementation
import time
import redis

class TokenBucketRateLimiter:
    def __init__(self, redis_client, key: str, capacity: int, refill_rate: float):
        self.r = redis_client
        self.key = key               # e.g. "ratelimit:resend_emails"
        self.capacity = capacity     # max burst (e.g. 100)
        self.refill_rate = refill_rate  # tokens per second (e.g. 10)

    def acquire(self) -> bool:
        """Returns True if a token was acquired, False if rate limited."""
        now = time.time()
        pipe = self.r.pipeline()

        # Lua script ensures atomic read-modify-write (no race conditions)
        lua_script = """
        local key = KEYS[1]
        local capacity = tonumber(ARGV[1])
        local refill_rate = tonumber(ARGV[2])
        local now = tonumber(ARGV[3])

        local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
        local tokens = tonumber(bucket[1]) or capacity
        local last_refill = tonumber(bucket[2]) or now

        -- Refill tokens based on time elapsed
        local elapsed = now - last_refill
        tokens = math.min(capacity, tokens + elapsed * refill_rate)

        if tokens >= 1 then
            tokens = tokens - 1
            redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
            redis.call('EXPIRE', key, 3600)
            return 1
        end
        return 0
        """
        result = self.r.eval(lua_script, 1, self.key, self.capacity, self.refill_rate, now)
        return bool(result)

# Usage inside a Celery task
limiter = TokenBucketRateLimiter(r, "ratelimit:resend", capacity=100, refill_rate=10)

@app.task(bind=True, max_retries=10)
def send_email_task(self, email, token):
    if not limiter.acquire():
        # Rate limited, retry after a short delay (not counted as failure)
        raise self.retry(countdown=1, max_retries=60)  # retry every 1s up to 60 times
    _do_send_email(email, token)

// rateLimiter.js, Redis token bucket implementation
const TOKEN_BUCKET_LUA = `
local key = KEYS[1]
local capacity = tonumber(ARGV[1])
local refill_rate = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1]) or capacity
local last_refill = tonumber(bucket[2]) or now

-- Refill tokens based on time elapsed
local elapsed = now - last_refill
tokens = math.min(capacity, tokens + elapsed * refill_rate)

if tokens >= 1 then
    tokens = tokens - 1
    redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
    redis.call('EXPIRE', key, 3600)
    return 1
end
return 0
`

export class TokenBucketRateLimiter {
  constructor(redis, key, capacity, refillRate) {
    this.redis = redis
    this.key = key                // e.g. "ratelimit:resend_emails"
    this.capacity = capacity      // max burst (e.g. 100)
    this.refillRate = refillRate  // tokens per second (e.g. 10)
  }

  // Returns true if a token was acquired, false if rate limited.
  async acquire() {
    const now = Date.now() / 1000
    const result = await this.redis.eval(TOKEN_BUCKET_LUA, {
      keys: [this.key],
      arguments: [String(this.capacity), String(this.refillRate), String(now)],
    })
    return result === 1
  }
}

// Usage inside a BullMQ worker
const limiter = new TokenBucketRateLimiter(redis, 'ratelimit:resend', 100, 10)

new Worker('email', async (job) => {
  if (!(await limiter.acquire())) {
    throw new Error('rate limited') // throwing re-queues with backoff
  }
  await doSendEmail(job.data)
}, { connection })

// BullMQ also ships a built-in limiter that PAUSES the worker instead of
// failing jobs, which is usually the better shape, no retry budget burned:
//   new Worker('email', fn, { connection, limiter: { max: 10, duration: 1000 } })
// Its scope is the queue, though, so a second unrelated queue hitting the
// same provider is not counted. A shared bucket like the one above is.

import type { RedisClientType } from 'redis'

// Two details decide whether this holds up in production:
//
//  1. `now` comes from the WORKER's clock, and ten pods do not agree to
//     the millisecond. If the refill has to be exact across machines,
//     take the time from Redis (`TIME`) and pass that instead.
//  2. It must stay ONE eval. Reading the bucket, computing the refill in
//     TypeScript, and writing it back is the chapter 08 lost-update race
//     wearing a different hat, and it leaks tokens under real load.
export class TokenBucketRateLimiter {
  constructor(
    private readonly redis: RedisClientType,
    private readonly key: string,        // e.g. "ratelimit:resend_emails"
    private readonly capacity: number,   // max burst (e.g. 100)
    private readonly refillRate: number, // tokens per second (e.g. 10)
  ) {}

  /** Returns true if a token was acquired, false if rate limited. */
  async acquire(): Promise<boolean> {
    const now = Date.now() / 1000
    const result = (await this.redis.eval(TOKEN_BUCKET_LUA, {
      keys: [this.key],
      arguments: [String(this.capacity), String(this.refillRate), String(now)],
    })) as number
    return result === 1
  }
}

@Component
public class TokenBucketRateLimiter {

    // Same Lua script, kept in src/main/resources/token_bucket.lua.
    // Spring caches its SHA and calls EVALSHA, so the body is not resent.
    private static final RedisScript<Long> SCRIPT = RedisScript.of(
        new ClassPathResource("token_bucket.lua"), Long.class);

    private final StringRedisTemplate redis;

    TokenBucketRateLimiter(StringRedisTemplate redis) {
        this.redis = redis;
    }

    /** Returns true if a token was acquired, false if rate limited. */
    public boolean acquire(String key, int capacity, double refillRate) {
        double now = System.currentTimeMillis() / 1000.0;
        Long result = redis.execute(SCRIPT, List.of(key),
            String.valueOf(capacity), String.valueOf(refillRate), String.valueOf(now));
        return Long.valueOf(1L).equals(result);
    }
}

// Usage inside the stream handler
if (!limiter.acquire("ratelimit:resend", 100, 10)) {
    // Rate limited: deliberately do NOT ack. The message stays pending
    // and gets reclaimed later, which is this broker's version of a retry.
    return;
}
doSendEmail(payload);

// Resilience4j ships a RateLimiter too, but its scope is ONE JVM. That is
// the wrong scope here: ten worker pods each allowing 100/s is 1000/s
// arriving at the provider. A shared limit has to live in Redis.

# sliding_window.py, Redis sorted-set sliding window
import time


class SlidingWindowLimiter:
    def __init__(self, redis_client, key: str, limit: int, window_seconds: float):
        self.r = redis_client
        self.key = key
        self.limit = limit            # max requests
        self.window = window_seconds  # time window

    def allow(self) -> bool:
        now_us = int(time.time() * 1_000_000)
        window_start = now_us - int(self.window * 1_000_000)

        pipe = self.r.pipeline()
        # Remove entries older than the window
        pipe.zremrangebyscore(self.key, 0, window_start)
        # Count entries in current window
        pipe.zcard(self.key)
        # Add current request timestamp as a member
        pipe.zadd(self.key, {str(now_us): now_us})
        pipe.expire(self.key, int(self.window * 2))
        _, count, _, _ = pipe.execute()

        return count < self.limit


# Usage inside a Celery task
@app.task(bind=True, max_retries=60)
def send_email_task(self, email, token):
    if not limiter.allow():
        raise self.retry(countdown=1)  # rate limited, not a failure
    _do_send_email(email, token)

// slidingWindow.js, Redis sorted-set sliding window
export class SlidingWindowLimiter {
  constructor(redis, key, limit, windowMs) {
    this.redis = redis
    this.key = key
    this.limit = limit        // max requests
    this.windowMs = windowMs  // time window
  }

  async allow() {
    const now = Date.now() * 1000 // microseconds, used as the score
    const windowStart = now - this.windowMs * 1000

    const [, count] = await this.redis
      .multi()
      // Remove entries older than the window
      .zRemRangeByScore(this.key, 0, windowStart)
      // Count entries in current window
      .zCard(this.key)
      // Add current request timestamp as a member
      .zAdd(this.key, { score: now, value: String(now) })
      .expire(this.key, Math.ceil((this.windowMs * 2) / 1000))
      .exec()

    return count < this.limit
  }
}

// Two behaviours of this algorithm that are easy to miss, and both are in
// the Go version above too:
//
//  1. The count is read BEFORE this request's own entry is added, so the
//     effective ceiling is limit + 1 in flight. Either compare against
//     `limit - 1`, or move the ZADD ahead of the ZCARD. Choose one on
//     purpose, because drifting between the two across services is how a
//     documented "100/s" quietly becomes 101/s at the provider.
//  2. An entry is added even when the request is REJECTED. A caller that
//     keeps hammering a limited endpoint therefore keeps its own window
//     permanently full and never recovers. If that is not what you want,
//     only ZADD on the allowed path.
export class SlidingWindowLimiter {
  constructor(
    private readonly redis: RedisClientType,
    private readonly key: string,
    private readonly limit: number,    // max requests
    private readonly windowMs: number, // time window
  ) {}

  async allow(): Promise<boolean> {
    const now = Date.now() * 1000
    const windowStart = now - this.windowMs * 1000

    const results = await this.redis
      .multi()
      .zRemRangeByScore(this.key, 0, windowStart)
      .zCard(this.key)
      .expire(this.key, Math.ceil((this.windowMs * 2) / 1000))
      .exec()

    const count = results[1] as number
    if (count >= this.limit) {
      return false // rejected, and deliberately NOT recorded
    }

    await this.redis.zAdd(this.key, { score: now, value: String(now) })
    return true
  }
}

@Component
public class SlidingWindowLimiter {
    private final StringRedisTemplate redis;

    SlidingWindowLimiter(StringRedisTemplate redis) {
        this.redis = redis;
    }

    public boolean allow(String key, int limit, Duration window) {
        long nowMicros = Instant.now().toEpochMilli() * 1_000L;
        long windowStart = nowMicros - window.toMillis() * 1_000L;

        // Remove entries older than the window
        redis.opsForZSet().removeRangeByScore(key, 0, windowStart);
        // Count entries in current window
        Long count = redis.opsForZSet().zCard(key);
        // Add current request timestamp as a member
        redis.opsForZSet().add(key, String.valueOf(nowMicros), nowMicros);
        redis.expire(key, window.multipliedBy(2));

        return count != null && count < limit;
    }
}

// Written plainly like this it is four round trips AND four separate
// moments: another worker can slip an entry in between the ZCARD and the
// ZADD. Pipeline it with executePipelined, or move the whole sequence
// into a Lua script the way the token bucket does. At low rates you will
// never notice; at the rates that make you reach for a limiter, you will.

# Python, Full metrics instrumentation for a worker process
from prometheus_client import (
    Counter, Histogram, Gauge, start_http_server
)
import time

# -- Define metrics ---------------------------------------------
TASK_TOTAL = Counter(
    'worker_tasks_total',
    'Total tasks processed',
    ['task_name', 'status']  # labels: task_name, status=success|failure
)

TASK_DURATION = Histogram(
    'worker_task_duration_seconds',
    'Task processing duration',
    ['task_name'],
    buckets=[.01, .05, .1, .5, 1, 5, 10, 30]  # histogram bucket boundaries
)

QUEUE_DEPTH = Gauge(
    'worker_queue_depth',
    'Current tasks waiting in queue',
    ['queue_name']
)

RETRY_COUNT = Counter(
    'worker_task_retries_total',
    'Number of task retries',
    ['task_name']
)

# -- Decorator to wrap any task with metrics --------------------
def track_metrics(task_name: str):
    def decorator(func):
        def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = func(*args, **kwargs)
                TASK_TOTAL.labels(task_name=task_name, status="success").inc()
                return result
            except Exception as e:
                TASK_TOTAL.labels(task_name=task_name, status="failure").inc()
                raise
            finally:
                TASK_DURATION.labels(task_name=task_name).observe(time.time() - start)
        return wrapper
    return decorator

# Start /metrics server on port 9090 (Prometheus scrapes this)
start_http_server(9090)

# -- Apply to task ----------------------------------------------
@app.task(bind=True, max_retries=5)
@track_metrics("send_verification_email")
def send_verification_email(self, user_id, email, token):
    if self.request.retries > 0:
        RETRY_COUNT.labels(task_name="send_verification_email").inc()
    _do_send_email(email, token)

// Node, Full metrics instrumentation for a worker process
import { Counter, Histogram, Gauge, register } from 'prom-client'
import express from 'express'

// -- Define metrics ---------------------------------------------
export const TASK_TOTAL = new Counter({
  name: 'worker_tasks_total',
  help: 'Total tasks processed',
  labelNames: ['task_name', 'status'], // status=success|failure
})

export const TASK_DURATION = new Histogram({
  name: 'worker_task_duration_seconds',
  help: 'Task processing duration',
  labelNames: ['task_name'],
  buckets: [0.01, 0.05, 0.1, 0.5, 1, 5, 10, 30], // histogram bucket boundaries
})

export const QUEUE_DEPTH = new Gauge({
  name: 'worker_queue_depth',
  help: 'Current tasks waiting in queue',
  labelNames: ['queue_name'],
})

export const RETRY_COUNT = new Counter({
  name: 'worker_task_retries_total',
  help: 'Number of task retries',
  labelNames: ['task_name'],
})

// -- Wrapper to give any processor metrics ----------------------
export function trackMetrics(taskName, processor) {
  return async (job) => {
    const end = TASK_DURATION.startTimer({ task_name: taskName })
    if (job.attemptsMade > 0) {
      RETRY_COUNT.inc({ task_name: taskName })
    }
    try {
      const result = await processor(job)
      TASK_TOTAL.inc({ task_name: taskName, status: 'success' })
      return result
    } catch (err) {
      TASK_TOTAL.inc({ task_name: taskName, status: 'failure' })
      throw err
    } finally {
      end() // records the duration on both the success and failure paths
    }
  }
}

// Start /metrics server on port 9090 (Prometheus scrapes this)
const metricsApp = express()
metricsApp.get('/metrics', async (_req, res) => {
  res.set('Content-Type', register.contentType)
  res.end(await register.metrics())
})
metricsApp.listen(9090)

// -- Apply to task ----------------------------------------------
new Worker('email', trackMetrics('send_verification_email', emailProcessor), { connection })

// QUEUE_DEPTH is the one no wrapper can fill in: it describes the QUEUE,
// not any single job, so it has to be polled on a timer.
setInterval(async () => {
  QUEUE_DEPTH.set({ queue_name: 'email' }, await emailQueue.getWaitingCount())
}, 15_000)

import type { Job } from 'bullmq'

// The Prometheus rule worth burning in before you add a single label:
// labels must be LOW cardinality. `task_name` has maybe twenty values,
// which is fine. A user id or a job id creates a new time series per
// user, and that is how people take their own Prometheus down. Pinning
// the label type instead of accepting an open Record is a cheap guard.
type TaskLabels = { task_name: string; status?: 'success' | 'failure' }

// The generics keep the wrapper invisible at the call site: what goes in
// is a processor, what comes out is the SAME processor type.
export function trackMetrics<T, R>(
  taskName: string,
  processor: (job: Job<T>) => Promise<R>,
): (job: Job<T>) => Promise<R> {
  return async (job: Job<T>): Promise<R> => {
    const labels: TaskLabels = { task_name: taskName }
    const end = TASK_DURATION.startTimer(labels)

    if (job.attemptsMade > 0) {
      RETRY_COUNT.inc(labels)
    }

    try {
      const result = await processor(job)
      TASK_TOTAL.inc({ ...labels, status: 'success' })
      return result
    } catch (err) {
      TASK_TOTAL.inc({ ...labels, status: 'failure' })
      throw err
    } finally {
      end() // records the duration on both paths
    }
  }
}

// Java, Full metrics instrumentation for a worker process.
// spring-boot-starter-actuator + micrometer-registry-prometheus give you
// the /actuator/prometheus endpoint for free; you define the metrics.
@Component
public class WorkerMetrics {
    private final MeterRegistry registry;

    WorkerMetrics(MeterRegistry registry, EmailQueue queue) {
        this.registry = registry;

        // A Gauge is POLLED, not incremented: Micrometer calls this
        // supplier at scrape time, which is the right shape for a value
        // that describes the queue rather than any one task.
        Gauge.builder("worker_queue_depth", queue::waitingCount)
             .tag("queue_name", "email")
             .register(registry);
    }

    /** Wraps any task body with the counter + histogram, like the decorator. */
    public <T> T track(String taskName, Supplier<T> body) {
        Timer.Sample sample = Timer.start(registry);
        try {
            T result = body.get();
            registry.counter("worker_tasks_total",
                "task_name", taskName, "status", "success").increment();
            return result;
        } catch (RuntimeException e) {
            registry.counter("worker_tasks_total",
                "task_name", taskName, "status", "failure").increment();
            throw e;
        } finally {
            // stop() in the finally block, so a failed task is timed too
            sample.stop(Timer.builder("worker_task_duration_seconds")
                .tag("task_name", taskName)
                .publishPercentileHistogram() // the histogram buckets
                .register(registry));
        }
    }
}

// application.yml, expose the scrape endpoint on the metrics port:
//   management.endpoints.web.exposure.include: prometheus
//   management.server.port: 9090