Pemanggilan get_or_create() pada Django ORM sering memicu django.db.utils.IntegrityError secara intermiten saat aplikasi menerima lonjakan trafik konkuren. Masalah ini lazim muncul pada worker Celery atau endpoint REST API berbeban tinggi, menghasilkan HTTP 500 atau kegagalan task tanpa penyebab yang langsung terlihat jelas dari log aplikasi standar.

Gejala: Intermittent 500 Error pada Beban Konkuren

Gejala umum dari bug ini meliputi:

  • Sistem mengembalikan HTTP 500 Internal Server Error secara acak pada endpoint idempotent (misal: webhook pembayaran, inisialisasi profil pengguna, atau klaim kupon).
  • Worker background (Celery/RQ) crash dengan traceback psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint atau padanan MySQL-nya IntegrityError: (1062, Duplicate entry).
  • Pengecekan database manual setelah insiden selalu menunjukkan data telah tersimpan rapi tanpa duplikasi fisik.

Anatomi Masalah: Race Window pada get_or_create()

Secara konseptual, get_or_create(**kwargs) beroperasi dengan alur kerja berikut:

  1. Eksekusi query SELECT berdasarkan kueri pencarian.
  2. Jika objek ditemukan, kembalikan (obj, False).
  3. Jika objek tidak ditemukan (DoesNotExist), masuki blok transaction.atomic() dan eksekusi INSERT.

Race condition terjadi di celah antara langkah 1 dan langkah 3. Pertimbangkan interaksi dua worker berikut:

Worker A: SELECT id FROM customer_wallet WHERE user_id = 42; -> Kosong
Worker B: SELECT id FROM customer_wallet WHERE user_id = 42; -> Kosong
Worker A: INSERT INTO customer_wallet (user_id, balance) VALUES (42, 0); -> Sukses (Commit)
Worker B: INSERT INTO customer_wallet (user_id, balance) VALUES (42, 0); -> Gagal: UniqueViolation

Database menolak INSERT dari Worker B untuk mempertahankan integritas referensial. Masalahnya berlanjut pada bagaimana Django merespons error tersebut.

Mengapa Savepoint Internal Django Tetap Melepaskan IntegrityError?

Django sebenarnya mengantisipasi benturan ini. Implementasi internal get_or_create membungkus INSERT ke dalam savepoint via transaction.atomic(). Saat IntegrityError tertangkap, Django melakukan rollback savepoint, lalu mencoba melakukan SELECT sekali lagi:

# Abstraksi logika internal Django get_or_create
try:
    return self.get(**lookup), False
except self.model.DoesNotExist:
    pass
try:
    with transaction.atomic(using=self.db):
        return self.create(**params), True
except IntegrityError:
    try:
        return self.get(**lookup), False
except self.model.DoesNotExist:
    pass
    raise  # <-- Exception bocor di sini

Exception tetap bocor ke aplikasi utama karena dua faktor teknis:

  1. Transaction Isolation Level (Repeatable Read): Pada PostgreSQL atau MySQL dengan isolasi REPEATABLE READ, Worker B terkunci dalam snapshot awal transaksinya. Saat Worker B mengeksekusi self.get(**lookup) kedua kali, snapshot tersebut belum melihat baris yang baru di-commit oleh Worker A. Hasilnya: self.get() kembali melempar DoesNotExist, dan Django mengeksekusi blok raise yang melepaskan IntegrityError asli.
  2. Unmatched Unique Constraints: Jika kolom pemicu IntegrityError bukan merupakan bagian langsung dari parameter pencarian (lookup), pemanggilan self.get(**lookup) kedua tetap gagal menemukan objek terkait, memaksa Django menaikkan exception.

Solusi Teknis dan Pola Perbaikan

1. Definisikan Constraint Valid di Level Database

Jangan mengandalkan logika aplikasi saja. Gunakan UniqueConstraint modern pada model Django untuk menjamin integritas di layer SQL.

# models.py
from django.db import models

class CustomerWallet(models.Model):
    user = models.OneToOneField('auth.User', on_delete=models.CASCADE, related_name='wallet')
    balance = models.DecimalField(max_digits=12, decimal_places=2, default=0)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        constraints = [
            models.UniqueConstraint(fields=['user'], name='unique_customer_wallet')
        ]

2. Implementasi Pola Retry Tangguh

Bungkus eksekusi di dalam fungsi pembantu dengan penanganan rollback dan pengulangan transaksional. Cara ini menangani latensi replika atau isolasi transaksi tanpa membiarkan 500 error lolos.

# services.py
import time
from django.db import IntegrityError, transaction
from .models import CustomerWallet

def get_or_create_wallet_safe(user, max_retries=3, delay=0.05):
    for attempt in range(max_retries):
        try:
            # Gunakan atomic block lokal per percobaan
            with transaction.atomic():
                return CustomerWallet.objects.get_or_create(user=user)
        except IntegrityError:
            if attempt == max_retries - 1:
                raise
            time.sleep(delay)  # Berikan jeda sejenak untuk resolusi snapshot
    raise IntegrityError("Gagal memperoleh atau membuat wallet setelah percobaan maksimum.")

3. Alternatif Database-Level: update_or_create dan ON CONFLICT

Pada Django 4.1 ke atas, metode bulk_create mendukung parameter update_conflicts=True dan unique_fields yang langsung memanfaatkan sintaks database ON CONFLICT DO UPDATE/NOTHING (PostgreSQL/SQLite) tanpa melalui race window aplikasi.

# Alternatif zero-race-window untuk multi-record insert
CustomerWallet.objects.bulk_create(
    [CustomerWallet(user=user, balance=0)],
    update_conflicts=True,
    unique_fields=['user'],
    update_fields=['balance']
)

Verifikasi: Concurrency Testing

Uji ketahanan endpoint atau fungsi terhadap konkurensi menggunakan concurrent.futures.ThreadPoolExecutor dalam unit test Django. Pengujian ini memastikan race condition tertangani tanpa melempar exception yang tidak tertangkap.

# tests/test_concurrency.py
from concurrent.futures import ThreadPoolExecutor
from django.contrib.auth.models import User
from django.test import TransactionTestCase
from myapp.services import get_or_create_wallet_safe
from myapp.models import CustomerWallet

class ConcurrentWalletCreationTest(TransactionTestCase):
    def test_concurrent_get_or_create(self):
        user = User.objects.create(username="concurrency_tester")
        num_workers = 8

        def worker_task():
            wallet, created = get_or_create_wallet_safe(user=user)
            return wallet.id

        with ThreadPoolExecutor(max_workers=num_workers) as executor:
            futures = [executor.submit(worker_task) for _ in range(num_workers)]
            wallet_ids = [f.result() for f in futures]

        # Semua worker harus mengembalikan ID wallet yang identik
        self.assertEqual(len(set(wallet_ids)), 1)
        # Hanya ada satu baris record di tabel
        self.assertEqual(CustomerWallet.objects.filter(user=user).count(), 1)

Gunakan TransactionTestCase alih-alih TestCase biasa agar transaksi commit secara nyata ke engine database pengujian, mereplikasi kondisi race window yang terjadi di production.