Tantangan Determinisme pada Uji Regresi Protokol P2P

Pengujian regresi pada sistem transfer data peer-to-peer (P2P)—seperti arsitektur BitTorrent, ed2k, atau Kademlia—sering menghasilkan flaky test di lingkungan continuous integration (CI). Masalah utama bersumber dari ketergantungan pada port eksternal, latensi I/O non-deterministik, dan penggunaan sleep() untuk menunggu sinkronisasi state antar peer. Runner CI yang mengalami CPU throttling dapat menyebabkan timer timeout terpicu sebelum transfer chunk selesai.

Untuk membangun uji regresi protokol P2P yang stabil, seluruh topologi swarm harus diisolasi ke dalam loopback interface lokal (127.0.0.1) dengan kontrol eksplisit atas event loop, latensi, drop rate, dan korupsi paket.

Arsitektur Mock Swarm Lokal

Mock swarm mengeliminasi kebutuhan DHT bootstrap eksternal atau server sentral publik. Komponen swarm direduksi menjadi tiga elemen utama yang berjalan dalam thread atau coroutine terpisah:

  • Tracker/Coordinator Minimal: Registry in-memory sederhana untuk mendistribusikan metadata peer tanpa overhead protokol routing tabel Kademlia.
  • Seeder Nodes: Instansi peer virtual yang memegang chunk payload valid beserta checksum/hash list (misal: Merkle tree atau MD4/SHA-256 chunk hashes).
  • Fault Injection Proxy: Lapisan perantara soket TCP/UDP yang dapat mencegat, menahan, merusak byte, atau membuang paket sebelum mencapai Leecher (Client under Test).

Desain Mock Socket dengan Fault Injection

Daripada menghubungkan client langsung ke peer, hubungkan koneksi melalui proxy loopback sederhana yang memiliki parameter drop_rate dan corrupt_rate. Berikut implementasi harness proxy berbasis soket Python standar:

import socket
import threading
import random
import hashlib

CHUNK_SIZE = 1024

class FaultyProxy(threading.Thread):
    def __init__(self, listen_port, target_port, drop_rate=0.0, corrupt_rate=0.0):
        super().__init__(daemon=True)
        self.listen_port = listen_port
        self.target_port = target_port
        self.drop_rate = drop_rate
        self.corrupt_rate = corrupt_rate
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(("127.0.0.1", self.listen_port))
        self.sock.listen(1)

    def run(self):
        client_conn, _ = self.sock.accept()
        target_conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        target_conn.connect(("127.0.0.1", self.target_port))

        t1 = threading.Thread(target=self._forward, args=(client_conn, target_conn), daemon=True)
        t2 = threading.Thread(target=self._forward, args=(target_conn, client_conn, True), daemon=True)
        t1.start()
        t2.start()
        t1.join()
        t2.join()

    def _forward(self, src, dst, apply_faults=False):
        while True:
            data = src.recv(CHUNK_SIZE)
            if not data:
                break
            if apply_faults:
                if random.random() < self.drop_rate:
                    continue  # Simulasi dropped packet / silent timeout
                if random.random() < self.corrupt_rate:
                    data = bytearray(data)
                    data[0] ^= 0xFF  # Balik bit pertama untuk korupsi chunk
                    data = bytes(data)
            dst.sendall(data)
        dst.close()
        src.close()

Simulasi Korupsi Chunk dan Re-request Logic

Protokol P2P membagi file ke dalam beberapa chunk terpisah yang diverifikasi secara independen menggunakan root hash atau chunk table. Saat proxy menginjeksi bitflip, protokol client harus mendeteksi ketidakcocokan hash sebelum menulis payload ke storage permanen.

Skenario uji regresi harus memastikan verifikasi hash end-to-end berjalan:

  1. Leecher meminta Chunk #0 dari Seeder melalui Proxy.
  2. Proxy merusak byte data pada Chunk #0.
  3. Leecher menghitung hash Chunk #0 yang diterima, mendeteksi ketidakcocokan terhadap metadata manifest.
  4. Leecher membuang buffer rusak dan mengirim kembali request (re-request) untuk Chunk #0.
  5. Proxy melewatkan paket kedua secara utuh.
  6. Transfer selesai; file lokal dicocokkan dengan root hash asli.

Eliminasi Timer Non-Deterministik di CI Runner

Penyebab utama kegagalan tes transfer file adalah penggunaan timer statis (contoh: wait_for_download(timeout=5)). Jika beban CPU runner meningkat, operasi I/O melambat dan tes gagal karena timeout palsu.

1. Virtual Clocks / Manual Time Progression

Ganti dependensi jam sistem (system wall clock) dengan event-driven ticks. Hindari memajukan waktu nyata; majukan state mesin client menggunakan loop eksplisit yang menunggu event tertentu (seperti ChunkDownloadedEvent atau VerificationFailedEvent).

2. Condition-based Polling over Sleep

Alih-alih memanggil time.sleep(), gunakan sinkronisasi berbasis thread primitive (threading.Event) atau polling interval kecil dengan batas maksimal berbasis kondisi logika internal.

def wait_for_condition(predicate, timeout_ms=5000, step_ms=50):
    import time
    deadline = time.monotonic() + (timeout_ms / 1000.0)
    while time.monotonic() < deadline:
        if predicate():
            return True
        time.sleep(step_ms / 1000.0)
    raise TimeoutError("Predicate condition failed within allocated timeout")

Contoh Test Harness End-to-End

Kode berikut mendemonstrasikan verifikasi transfer file lokal menggunakan seeder in-memory, proxy dengan fault injection, dan leecher yang memvalidasi integritas SHA-256 tanpa akses jaringan luar.

import socket
import threading
import hashlib
from io import BytesIO

# 1. Mock Seeder
class MockSeeder(threading.Thread):
    def __init__(self, port, data):
        super().__init__(daemon=True)
        self.port = port
        self.data = data
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(("127.0.0.1", self.port))
        self.sock.listen(1)

    def run(self):
        conn, _ = self.sock.accept()
        conn.sendall(self.data)
        conn.close()
        self.sock.close()

# 2. Test Execution
def test_p2p_transfer_integrity():
    payload = b"Deterministic-P2P-Chunk-Payload-1234567890" * 32
    expected_hash = hashlib.sha256(payload).hexdigest()

    seeder_port = 9001
    proxy_port = 9002

    seeder = MockSeeder(seeder_port, payload)
    seeder.start()

    # Inisialisasi proxy tanpa fault untuk baseline integritas
    proxy = FaultyProxy(proxy_port, seeder_port, drop_rate=0.0, corrupt_rate=0.0)
    proxy.start()

    # Leecher client sederhana
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect(("127.0.0.1", proxy_port))

    buffer = BytesIO()
    while True:
        chunk = client.recv(1024)
        if not chunk:
            break
        buffer.write(chunk)
    client.close()

    received_data = buffer.getvalue()
    received_hash = hashlib.sha256(received_data).hexdigest()

    assert received_hash == expected_hash, f"Hash mismatch: {received_hash} != {expected_hash}"

if __name__ == "__main__":
    test_p2p_transfer_integrity()
    print("Test regression passed: data transfer verified deterministically.")

Checklist Mitigasi Flaky Test P2P pada CI

  • Binding Port Dinamis: Jangan gunakan port hardcoded saat menjalankan tes secara paralel. Ikat soket ke port 0 agar sistem operasi memberikan port loopback yang tersedia secara acak.
  • Isolasi Storage: Arahkan semua penulisan chunk ke virtual filesystem berbasis RAM (seperti tmpfs atau /dev/shm di Linux runner) untuk menghindari lag disk I/O.
  • Log Dump on Failure: Simpan buffer paket jaringan dan status state machine internal hanya saat tes assertion gagal, menghindari overhead penulisan disk log pada tes yang berhasil.