Idempotency

Redis idempotency, locks, queues, and retry-safe side effects.

Idempotency

Copy-Paste Example

Idempotent queue worker:

import { loadService } from '@hile/core'
import redisService from '@hile/ioredis'
import { RedisIdempotency, stableHash } from '@hile/redis-idempotency'
import { RedisStreamQueue, defineQueue } from '@hile/redis-stream-queue'

type EmailJob = { tenantId: string; userId: string; template: 'welcome' }

const emailQueue = defineQueue<EmailJob>('email')

const redis = await loadService(redisService)
const queue = new RedisStreamQueue(redis, { prefix: 'app:' })
const idempotency = new RedisIdempotency(redis)

queue.worker(emailQueue, async (job) => {
  await idempotency.run(
    `idem:email:${job.data.tenantId}:${job.jobId ?? job.id}`,
    () => sendEmail(job.data),
    {
      lockTtl: 60_000,
      resultTtl: 86_400_000,
      fingerprint: stableHash(job.data),
    },
  )
}, {
  group: 'email-workers',
  consumer: process.env.HOSTNAME ?? `${process.pid}`,
  concurrency: 8,
}).start()

Complete Example

// src/services/email-worker.boot.ts
import { defineService, loadService } from '@hile/core'
import redisService from '@hile/ioredis'
import { RedisIdempotency, stableHash } from '@hile/redis-idempotency'
import { RedisStreamQueue, defineQueue } from '@hile/redis-stream-queue'

type EmailPayload = {
  tenantId: string
  userId: string
  template: 'welcome'
}

const emailQueue = defineQueue<EmailPayload>('email')

export default defineService('email.worker', async (shutdown) => {
  const redis = await loadService(redisService)
  const queue = new RedisStreamQueue(redis, { prefix: 'app:' })
  const idempotency = new RedisIdempotency(redis)

  const worker = queue.worker(emailQueue, async (job) => {
    await idempotency.run(
      `idem:email:${job.data.tenantId}:${job.jobId ?? job.id}`,
      () => sendEmail(job.data),
      {
        lockTtl: 60_000,
        resultTtl: 86_400_000,
        fingerprint: stableHash(job.data),
      },
    )
  }, {
    group: 'email-workers',
    consumer: process.env.HOSTNAME ?? `${process.pid}`,
    concurrency: 8,
    claimIdle: 60_000,
  })

  worker.start()
  shutdown(() => worker.stop())

  return { queue, worker }
})

Enqueue:

await queue.add(emailQueue, {
  tenantId: 't1',
  userId: 'u1',
  template: 'welcome',
}, {
  jobId: 'welcome:t1:u1',
  maxAttempts: 5,
  backoff: { type: 'exponential', baseMs: 1_000, maxMs: 60_000 },
})

Runtime And Lifecycle Notes

• RedisLock.withLock() asserts ownership before returning a successful callback result.
• tryLock() returns undefined if the key is already locked.
• RedisIdempotency.run() stores IN_FLIGHT and DONE states in Redis and uses Redis locks for ownership.
• Idempotency requires a stable fingerprint.
• stableHash() is for plain DTOs and rejects unsupported shapes.
• Rate limit algorithms: fixed-window, sliding-window, token-bucket.
• dryRun rate limit checks do not mutate Redis.
• Queue workers use Redis Streams consumer groups, delayed sorted sets, pending claim recovery, retry backoff, and DLQ streams.
• Queue payload schema can expose parse() or safeParse().

Anti-Patterns

• Random UUID per retry as idempotency key.
• Returning Date, BigInt, or class instances from idempotent functions without a resultCodec.
• Using stream() or RPC retries without idempotency around irreversible handlers.
• Ignoring DLQ after max attempts.

Verification Checklist

• Lock TTL is longer than normal critical-section time.
• Fencing is enabled when stale owners can write to external resources.
• Idempotency keys use business identifiers.
• Queue handlers are idempotent or wrapped in idempotency.
• Rate limit Retry-After is exposed in seconds for HTTP.
• DLQ is monitored.