Server sesekali mengalami galat 500, koneksi terputus di tengah respons, atau proses restart ketika permintaan kita sedang diproses. Galat-galat tersebut sifatnya sementara: permintaan yang sama besar kemungkinan berhasil bila diulang beberapa saat kemudian. Namun mengulang seketika justru menumpukkan permintaan pada server yang sedang bergulat dengan masalah, terutama bila puluhan klien melakukan hal yang sama. Pada bagian ini kita menerapkan exponential backoff dengan jitter di src/retry.py.

Kebijakan Percobaan Ulang #

Seluruh parameter retry dikumpulkan dalam satu dataclass agar mudah dibaca dari config/pipeline.yaml:

python 
@dataclass
class RetryPolicy:
    """Parameter percobaan ulang yang dipakai ApiClient.

    max_retries: jumlah percobaan ulang setelah permintaan pertama.
    backoff_base: durasi tunggu awal dalam detik.
    backoff_cap: batas atas durasi tunggu agar pipeline tidak menggantung.
    jitter: fraksi pengacakan durasi (0.25 berarti +-25%).
    max_retry_after: nilai Retry-After terbesar yang mau kita hormati.
    """

    max_retries: int = 3
    backoff_base: float = 0.5
    backoff_cap: float = 30.0
    jitter: float = 0.25
    max_retry_after: float = 60.0

Perhitungan durasinya berada di fungsi murni backoff_delay:

python 
def backoff_delay(attempt, policy=None, rng=random.random):
    """Durasi tunggu sebelum percobaan ulang ke-``attempt`` (mulai dari 0).

    Durasi dasar tumbuh eksponensial (base * 2^attempt) sampai ``backoff_cap``,
    lalu dikalikan faktor jitter agar banyak klien tidak mencoba ulang pada
    detik yang sama.
    """
    policy = policy or RetryPolicy()
    dasar = min(policy.backoff_base * (2 ** attempt), policy.backoff_cap)
    faktor = 1.0 + policy.jitter * (2.0 * rng() - 1.0)
    return dasar * faktor

Mari kita telaah baris-baris kode di atas:

  1. Baris 34 (min(base * 2**attempt, cap)): Durasi dasar berlipat dua pada setiap kegagalan, 0,5 detik lalu 1, 2, 4, 8 detik, dan berhenti di backoff_cap 30 detik.
  2. Baris 35 (faktor = 1.0 + jitter * ...): Jitter menggeser durasi secara acak di kisaran tertentu; dengan jitter=0.25, durasi bergerak antara 75% hingga 125% dari durasi dasar.
  3. Baris 36 (return dasar * faktor): Fungsi tidak menyentuh time.sleep sama sekali; fungsi hanya menghitung durasi, sehingga mudah diuji tanpa menunggu waktu nyata.
  4. Parameter rng: Pembangkit acak bisa disuntik, dan tes memakai rng konstan untuk memeriksa batas atas serta batas bawah durasi.

Dengan nilai bawaan tersebut, tabel durasi tunggu untuk empat percobaan pertama terlihat sebagai berikut:

Percobaan ulangDurasi dasarKisaran dengan jitter 25%
ke-10,5 detik0,38 - 0,63 detik
ke-21,0 detik0,75 - 1,25 detik
ke-32,0 detik1,50 - 2,50 detik
ke-44,0 detik3,00 - 5,00 detik

Jitter ada untuk kasus banyak klien: bila seratus pipeline mengalami galat yang sama dan semuanya menunggu tepat 2,0 detik, seratus permintaan tiba kembali pada milidetik yang sama dan mengulang kemacetan. Pengacakan kecil meratakan beban pemulihan.

Pulih dari Galat 500 #

Jalankan server dengan --fail-first 2 sehingga dua permintaan pertama dibalas 500, lalu perhatikan klien bekerja:

bash 
python3 - <<'PY'
import logging

from src.client import ApiClient
from src.retry import RetryPolicy

logging.basicConfig(level=logging.WARNING,
                    format="%(levelname)s %(name)s: %(message)s")

policy = RetryPolicy(max_retries=3, backoff_base=0.2, backoff_cap=5.0, jitter=0.25)
klien = ApiClient("http://127.0.0.1:8765", timeout=5.0, policy=policy)
payload = klien.get_json("/api/v1/health")
print("Respons akhir:", payload)
PY

Output yang Dihasilkan:

text 
WARNING ingestion.client: percobaan 1 gagal (HTTP 500); tunggu 0.21 detik lalu ulangi
WARNING ingestion.client: percobaan 2 gagal (HTTP 500); tunggu 0.37 detik lalu ulangi
Respons akhir: {'status': 'ok'}

Durasi tunggu 0,21 dan 0,37 detik adalah durasi dasar 0,2 detik yang telah dikenai jitter, sehingga angka pada mesin Anda sedikit berbeda setiap eksekusi. Percobaan ketiga berhasil dan pemanggil menerima payload seolah tidak pernah ada gangguan.

Ketika Anggaran Habis #

Server dengan --fail-first 9 tidak pernah pulih dalam jangka percobaan kita. Klien menyerah dengan galat yang jelas alasan dan jumlah percobaannya:

bash 
python3 - <<'PY'
import logging

from src.client import ApiClient
from src.retry import RetryPolicy

logging.basicConfig(level=logging.WARNING,
                    format="%(levelname)s %(name)s: %(message)s")

policy = RetryPolicy(max_retries=3, backoff_base=0.2, backoff_cap=5.0, jitter=0.25)
klien = ApiClient("http://127.0.0.1:8765", timeout=5.0, policy=policy)
payload = klien.get_json("/api/v1/health")
print("Respons akhir:", payload)
PY

Output yang Dihasilkan:

text 
WARNING ingestion.client: percobaan 1 gagal (HTTP 500); tunggu 0.22 detik lalu ulangi
WARNING ingestion.client: percobaan 2 gagal (HTTP 500); tunggu 0.36 detik lalu ulangi
WARNING ingestion.client: percobaan 3 gagal (HTTP 500); tunggu 0.93 detik lalu ulangi
Traceback (most recent call last):
  File "<stdin>", line 11, in <module>
  File "/home/z/my-project/tutorial/content/data-engineering/proyek/pipeline-ingestion-api/src/client.py", line 49, in get_json
    self._tunggu_atau_gagal(attempt, url, f"HTTP {respons.status_code}",
  File "/home/z/my-project/tutorial/content/data-engineering/proyek/pipeline-ingestion-api/src/client.py", line 70, in _tunggu_atau_gagal
    raise TransientApiError(
src.client.TransientApiError: HTTP 500 pada http://127.0.0.1:8765/api/v1/health; batas 3 percobaan ulang tercapai

Tiga peringatan pertama memperlihatkan percobaan yang makin jarang karena durasi tunggu tumbuh; empat baris terakhir adalah traceback lengkap dari galat final. TransientApiError sengaja dibedakan dari ApiError permanen: penjadwal pipeline boleh menjalankan ulang seluruh run nanti, sedangkan galat permanen menuntut perbaikan kode atau konfigurasi terlebih dahulu. Pola kegagalan semacam ini dibahas pada chapter Menangani Kegagalan Ingestion.

Uji Retry #

Tes di tests/test_retry.py memeriksa kurva durasi dan perilaku nyata terhadap server flaky:

python 
def test_backoff_tumbuh_eksponensial_lalu_berhenti_di_cap():
    policy = kebijakan()
    delays = [backoff_delay(attempt, policy) for attempt in range(8)]
    assert delays == [0.5, 1.0, 2.0, 4.0, 8.0, 16.0, 30.0, 30.0]


def test_jitter_ada_di_kisaran_terbatas():
    policy = kebijakan(jitter=0.5)
    kecil = backoff_delay(0, policy, rng=lambda: 0.0)
    besar = backoff_delay(0, policy, rng=lambda: 1.0)
    # jitter 0.5 berarti durasi bergerak di kisaran +-50% dari durasi dasar.
    assert kecil == pytest.approx(0.5 * 0.5)
    assert besar == pytest.approx(0.5 * 1.5)

Tes pertama memakai jitter=0.0 sehingga deret durasi bisa dibandingkan persis, termasuk berhenti di nilai cap. Tes kedua menyuntik rng konstan pada dua ujungnya untuk membuktikan jitter tidak pernah keluar dari kisaran yang dijanjikan. Tes ketiga di berkas yang sama memanfaatkan status_log server flaky untuk membuktikan urutan kejadian [500, 500, 200] yang sesungguhnya terjadi di kabel jaringan.

bash 
python3 -m pytest tests/test_retry.py -q

Output yang Dihasilkan:

text 
....                                                                     [100%]
4 passed in 1.13s

Klien kita kini tahan banting terhadap tiga gangguan sekaligus: kuota, galat server, dan gangguan jaringan. Namun semua ketangguhan tersebut percuma bila payload yang diterima berubah bentuk tanpa sepengetahuan kita; bagian berikutnya memasang pagar schema sebelum data dipercaya. Perilaku time.sleep yang dipakai penundaan terdokumentasi di dokumentasi modul time Python .