Skip to content

Job Uniqueness Patterns — Complete Guide

DodaTech Updated 2026-06-28 6 min read

In this tutorial, you will learn about Job Uniqueness Patterns. We cover key concepts, practical examples, and best practices to help you master this topic.

Ensure exactly-once job processing with uniqueness constraints, idempotency keys, database unique constraints, and Redis-based deduplication for reliable queues.

What You Learn

You will learn how to enforce job uniqueness at the queue level, use database constraints for uniqueness, implement time-window uniqueness, and combine deduplication with idempotency.

Why It Matters

Duplicate jobs cause duplicate emails, double payments, and redundant processing. Uniqueness guarantees that the same logical operation executes only once, even if submitted multiple times.

Real-World Use

DodaTech's payment system enforces uniqueness with database-level constraints on transaction IDs. Webhook handlers use idempotency keys with Redis SETNX. Together they prevent duplicate payments.

Uniqueness Architecture

flowchart LR
    A[Submit Job] --> B{Unique Check}
    B -->|New| C[Enqueue]
    B -->|Duplicate| D[Reject]
    C --> E[Process]
    E --> F{Idempotent?}
    F -->|Yes| G[Safe to retry]
    F -->|No| H[Exactly once]
    D --> I[Return existing result]

Redis SETNX Uniqueness

import redis
import json
import time
import uuid

r = redis.Redis()

class UniqueJobQueue:
    def __init__(self, ttl=86400):
        self.ttl = ttl

    def enqueue_unique(self, queue, uniqueness_key, job_data):
        lock_key = f'unique:{queue}:{uniqueness_key}'
        acquired = r.setnx(lock_key, '1')
        if not acquired:
            print(f"Duplicate blocked: {uniqueness_key}")
            return None

        r.expire(lock_key, self.ttl)
        job_id = str(uuid.uuid4())
        job_data['_unique_key'] = uniqueness_key
        job_data['_job_id'] = job_id
        r.lpush(queue, json.dumps(job_data))
        print(f"Enqueued unique job: {uniqueness_key}")
        return job_id

    def is_processed(self, queue, uniqueness_key):
        return r.exists(f'unique:{queue}:{uniqueness_key}')

    def release(self, queue, uniqueness_key):
        r.delete(f'unique:{queue}:{uniqueness_key}')

uq = UniqueJobQueue()
uq.enqueue_unique('email_queue', 'welcome-user-123',
                  {'to': 'user@example.com', 'template': 'welcome'})
uq.enqueue_unique('email_queue', 'welcome-user-123',
                  {'to': 'user@example.com', 'template': 'welcome'})

Expected output:

Enqueued unique job: welcome-user-123
Duplicate blocked: welcome-user-123

Database Unique Constraint

import sqlite3
import time
import json

class DBUniqueQueue:
    def __init__(self, db_path=':memory:'):
        self.conn = sqlite3.connect(db_path)
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS unique_jobs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                job_type TEXT NOT NULL,
                uniqueness_key TEXT NOT NULL,
                status TEXT DEFAULT 'pending',
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                UNIQUE(job_type, uniqueness_key)
            )
        ''')
        self.conn.commit()

    def submit_unique(self, job_type, uniqueness_key, payload):
        try:
            self.conn.execute(
                'INSERT INTO unique_jobs (job_type, uniqueness_key) VALUES (?, ?)',
                (job_type, uniqueness_key)
            )
            self.conn.commit()
            job_id = self.conn.execute(
                'SELECT id FROM unique_jobs WHERE job_type = ? AND uniqueness_key = ?',
                (job_type, uniqueness_key)
            ).fetchone()[0]
            print(f"Created unique job: {job_type}/{uniqueness_key} (id={job_id})")
            return job_id
        except sqlite3.IntegrityError:
            existing = self.conn.execute(
                'SELECT id, status FROM unique_jobs WHERE job_type = ? AND uniqueness_key = ?',
                (job_type, uniqueness_key)
            ).fetchone()
            print(f"Duplicate rejected: {job_type}/{uniqueness_key} (status={existing[1]})")
            return None

    def mark_completed(self, job_id):
        self.conn.execute(
            'UPDATE unique_jobs SET status = ? WHERE id = ?',
            ('completed', job_id)
        )
        self.conn.commit()

    def get_status(self, job_type, uniqueness_key):
        cursor = self.conn.execute(
            'SELECT status FROM unique_jobs WHERE job_type = ? AND uniqueness_key = ?',
            (job_type, uniqueness_key)
        )
        row = cursor.fetchone()
        return row[0] if row else 'not_found'

dbq = DBUniqueQueue()
dbq.submit_unique('payment', 'txn-20260628-001', {'amount': 100})
dbq.submit_unique('payment', 'txn-20260628-001', {'amount': 100})

Expected output:

Created unique job: payment/txn-20260628-001 (id=1)
Duplicate rejected: payment/txn-20260628-001 (status=pending)

Time-Window Uniqueness

import redis
import json
import time

r = redis.Redis()

class TimeWindowUniqueQueue:
    def __init__(self, window_seconds=3600):
        self.window = window_seconds

    def enqueue_in_window(self, queue, namespace, resource_id, job_data):
        key = f'window_unique:{namespace}:{resource_id}'
        exists = r.exists(key)
        if exists:
            ttl = r.ttl(key)
            print(f"Duplicate in window: {namespace}/{resource_id} (TTL: {ttl}s)")
            return False

        r.setex(key, self.window, '1')
        r.lpush(queue, json.dumps(job_data))
        print(f"Enqueued: {namespace}/{resource_id}")
        return True

    def queue_depth(self, queue):
        return r.llen(queue)

twq = TimeWindowUniqueQueue(window_seconds=60)
twq.enqueue_in_window('scans', 'file_scan', 'doc-123',
                      {'file': 'doc-123', 'scanner': 'av'})
twq.enqueue_in_window('scans', 'file_scan', 'doc-123',
                      {'file': 'doc-123', 'scanner': 'av'})
time.sleep(61)
twq.enqueue_in_window('scans', 'file_scan', 'doc-123',
                      {'file': 'doc-123', 'scanner': 'av'})

Expected output:

Enqueued: file_scan/doc-123
Duplicate in window: file_scan/doc-123 (TTL: 58s)
Enqueued: file_scan/doc-123

Hash-Based Uniqueness

import hashlib
import json
import redis

r = redis.Redis()

class HashBasedUniqueness:
    def __init__(self, ttl=86400):
        self.ttl = ttl

    def compute_hash(self, job_data):
        serialized = json.dumps(job_data, sort_keys=True)
        return hashlib.sha256(serialized.encode()).hexdigest()

    def enqueue_if_unique(self, queue, job_data, namespace='job'):
        content_hash = self.compute_hash(job_data)
        dedup_key = f'hash_unique:{namespace}:{content_hash}'

        acquired = r.setnx(dedup_key, '1')
        if not acquired:
            print(f"Duplicate hash: {content_hash[:16]}...")
            return None

        r.expire(dedup_key, self.ttl)
        job_data['_hash'] = content_hash
        r.lpush(queue, json.dumps(job_data))
        print(f"Enqueued hash: {content_hash[:16]}...")
        return content_hash

hbu = HashBasedUniqueness()
job = {'type': 'report', 'params': {'date': '2026-06-28', 'format': 'pdf'}}
hbu.enqueue_if_unique('reports', job)
hbu.enqueue_if_unique('reports', job)

Expected output:

Enqueued hash: a1b2c3d4e5f6a7b8...
Duplicate hash: a1b2c3d4e5f6a7b8...

Common Mistakes

1. Uniqueness Without Expiration

Unique keys without TTL grow forever and leak memory. Always set TTL based on the maximum expected retry window.

2. Key Too Narrow

Uniqueness on job ID alone lets different jobs with same ID be blocked. Include queue name, job type, and content fingerprint in the key.

3. Assuming Queue-Level Uniqueness

Queues do not enforce uniqueness by default. You must implement it at the application layer using Redis, database, or distributed locks.

4. Ignoring Race Conditions

Two concurrent submissions may both pass the uniqueness check. Use atomic operations (SETNX, INSERT with unique constraint) not check-then-set patterns.

5. No Fallback on Store Failure

If Redis is down, uniqueness checks fail open (all jobs accepted) or fail closed (all jobs rejected). Design based on your safety requirements.

Practice Questions

1. What is the difference between uniqueness and idempotency?

Uniqueness prevents duplicate submissions. Idempotency ensures safe processing of duplicates. Uniqueness is proactive, idempotency is defensive.

2. How does SETNX guarantee atomic uniqueness checks?

SETNX only sets a key if it does not exist. The atomic check-and-set prevents race conditions between concurrent submissions.

3. When should you use time-window uniqueness?

When the same operation should not repeat within a window but should be allowed after (e.g., rate-limited email sending, periodic cache refresh).

4. Why include namespace in uniqueness keys?

Different job types may share the same key (e.g., user ID 123). Namespacing prevents false duplicates across different job types.

Challenge

Build a uniqueness system for a payment processor that supports: unique transaction IDs with database constraints, retry idempotency within 24 hours, concurrent submission safety, and status reporting for duplicate queries.

FAQ

Can I use Redis SET for uniqueness?

SETNX is atomic. Regular SET with EXISTS check is not atomic and has race conditions. Always use SETNX or Lua scripts for uniqueness.

What happens to uniqueness keys after job completion?

Keys can be deleted immediately or kept with TTL for auditing. Keep them at least as long as the retry window.

Does SQS support built-in deduplication?

SQS FIFO queues support exactly-once delivery within a 5-minute deduplication window. Standard SQS does not. Use application-level dedup for longer windows.

How do I handle uniqueness across multiple regions?

Use globally unique identifiers (UUID v4, ULID). Include region or data center in the uniqueness key for cross-region coordination.

Can uniqueness cause job loss?

If the uniqueness key is too broad, different jobs are incorrectly blocked. Scope keys precisely to job identity, not to shared attributes.

Mini Project: Uniqueness System

import redis
import json
import time
import hashlib

r = redis.Redis()

class UniquenessSystem:
    def __init__(self, default_ttl=86400):
        self.default_ttl = default_ttl

    def submit(self, queue, job_type, identifier, job_data):
        unique_key = f"unique:{queue}:{job_type}:{identifier}"
        acquired = r.setnx(unique_key, json.dumps({'status': 'pending', 'created': time.time()}))
        if not acquired:
            existing = r.get(unique_key)
            print(f"Duplicate: {job_type}/{identifier} - {existing.decode()}")
            return None
        r.expire(unique_key, self.default_ttl)
        r.lpush(queue, json.dumps({**job_data, '_uid': identifier}))
        return identifier

    def complete(self, queue, job_type, identifier):
        key = f"unique:{queue}:{job_type}:{identifier}"
        r.setex(key, self.default_ttl,
                json.dumps({'status': 'completed', 'completed_at': time.time()}))

    def status(self, queue, job_type, identifier):
        key = f"unique:{queue}:{job_type}:{identifier}"
        data = r.get(key)
        if data:
            return json.loads(data)
        return None

us = UniquenessSystem()
us.submit('payments', 'payment', 'txn_001', {'amount': 50})
us.submit('payments', 'payment', 'txn_001', {'amount': 50})
us.complete('payments', 'payment', 'txn_001')
print(us.status('payments', 'payment', 'txn_001'))

Expected output:

Duplicate: payment/txn_001 - {"status": "pending", "created": ...}
{'status': 'completed', 'completed_at': ...}

What's Next

Now that you understand job uniqueness, explore job chaining for sequential execution, then learn about job DAG workflows for complex Orchestration.

← Previous Recurring Job Scheduling with Cron Next → Job Chaining and Sequential Execution

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Background Jobs