Integrasi feed General Transit Feed Specification Realtime (GTFS-RT) untuk pelacakan armada kereta sering kali menghasilkan anomali visual: kereta melompat mundur (rubberbanding), rute membeku sejenak, atau data posisi terbaru tertimpa oleh data lama. Masalah ini bukan berasal dari kesalahan protokol GTFS-RT, melainkan karakteristik jaringan seluler dan variasi perangkat keras di lapangan yang memicu out-of-order events, jitter, serta clock drift.

Untuk membangun sistem ingestion yang konsisten tanpa membebani performa pemrosesan, backend ingest harus memvalidasi data telemetri sebelum mendistribusikannya ke downstream consumer atau visualizer WebSocket.

Akar Masalah: Jitter Seluler dan Asinkronisasi Jam

Saat unit onboard (OBU) kereta mengirim pembaruan posisi melalui jaringan seluler di sepanjang lintasan rel, tiga kendala utama muncul secara reguler:

  • Out-of-Order Delivery: Retry TCP/HTTP di area sinyal lemah menyebabkan paket lama yang tertunda tiba di server ingestion setelah paket yang lebih baru berhasil diproses.
  • Clock Drift: RTC (Real-Time Clock) pada OBU kereta bergeser atau tidak sinkron dengan server ingest. Jika NTP OBU bermasalah atau sinyal GPS lock terputus sementara, timestamp yang dikirim bisa berada puluhan detik di masa lalu atau masa depan.
  • Payload Redundancy: Protokol polling GTFS-RT konvensional mengembalikan seluruh snapshot entitas aktif. Backend yang menarik feed setiap beberapa detik akan berulang kali menerima data titik koordinat yang sama persis untuk armada yang sedang berhenti.

Desain Kontrak Ingestion dan State Tracking

Menangani edge case ini memerlukan validasi multi-tahap pada layer ingest sebelum mutasi state dilakukan pada database atau message bus:

1. Deduplikasi dengan Composite Key

Gunakan format identitas idempotensi unik untuk setiap pembacaan telemetri:

composite_key = f"{entity.vehicle.trip.trip_id}:{entity.vehicle.vehicle.id}:{entity.vehicle.timestamp}"

Simpan key ini di memory cache (seperti Redis atau in-memory LRU) dengan TTL pendek (misal 60 detik). Paket dengan key yang identik langsung dibuang untuk menghemat alokasi memori downstream worker.

2. Validasi Batas Waktu (Clock Drift Guard)

Tetapkan window toleransi perbedaan antara timestamp paket ($T_{event}$) dan waktu lokal server ingest ($T_{server}$):

  • Future Drift: Jika $T_{event} > T_{server} + \Delta_{max\_future}$ (misal 5 detik), tolak payload atau log anomali. Data di masa depan berisiko memblokir seluruh update valid berikutnya.
  • Stale Latency: Jika $T_{event} < T_{server} - \Delta_{max\_past}$ (misal 120 detik), buang payload karena data sudah tidak relevan untuk visualisasi posisi real-time.

3. Monotonic Timestamp Gate (Stale Drop Logic)

Simpan nilai last_processed_timestamp untuk setiap vehicle_id. Jika paket baru yang lolos deduplikasi memiliki $T_{event} \le T_{last\_processed}$, paket tersebut merupakan paket out-of-order yang terlambat tiba. Paket ini harus langsung di-drop untuk menghindari pergerakan mundur.

Implementasi Sliding Jitter Buffer dan Telemetry Processor

Jika downstream client membutuhkan urutan posisi yang benar-benar mulus tanpa langsung membuang paket yang tiba selisih sepersekian detik, gunakan Sliding Jitter Buffer berbasis min-heap. Buffer ini menahan paket selama rentang waktu konstan (misal 1000–2000 ms) agar paket yang tertunda dapat disusun ulang sebelum diteruskan ke broadcast layer.

Berikut adalah implementasi Python minimal menggunakan asyncio dan struktur data efisien untuk memproses stream data GTFS-RT:

import asyncio
import heapq
import time
from dataclasses import dataclass, field
from typing import Dict, Optional, Tuple

@dataclass(order=True)
class TelemetryPacket:
    timestamp: int
    vehicle_id: str = field(compare=False)
    latitude: float = field(compare=False)
    longitude: float = field(compare=False)
    received_at: float = field(compare=False, default_factory=time.monotonic)

class GTFSIngestionPipeline:
    def __init__(self, buffer_delay_sec: float = 1.5, max_future_drift_sec: int = 5):
        self.buffer_delay = buffer_delay_sec
        self.max_future_drift = max_future_drift_sec
        self.jitter_buffer: list[TelemetryPacket] = []
        self.last_applied_timestamps: Dict[str, int] = {}
        self.seen_signatures: set[Tuple[str, int]] = set()

    def ingest(self, vehicle_id: str, timestamp: int, lat: float, lon: float) -> bool:
        current_wall_time = int(time.time())

        # 1. Validasi Clock Drift (Masa depan)
        if timestamp > current_wall_time + self.max_future_drift:
            # Anomali jam pada OBU kendaraan
            return False

        # 2. Deduplikasi Cepat
        sig = (vehicle_id, timestamp)
        if sig in self.seen_signatures:
            return False

        # 3. Monotonic Check terhadap State Terakhir yang Sudah Disalurkan
        last_ts = self.last_applied_timestamps.get(vehicle_id, 0)
        if timestamp <= last_ts:
            # Stale drop: data lebih usang daripada posisi yang sudah dikonsumsi
            return False

        # Masukkan ke Priority Queue (Urut berdasarkan event timestamp)
        self.seen_signatures.add(sig)
        heapq.heappush(self.jitter_buffer, TelemetryPacket(timestamp, vehicle_id, lat, lon))
        return True

    async def consumer_loop(self, output_callback):
        """Mengalirkan data terurut ke downstream consumer setelah melewati jendela jitter."""
        while True:
            now = time.monotonic()
            while self.jitter_buffer and (now - self.jitter_buffer[0].received_at) >= self.buffer_delay:
                packet = heapq.heappop(self.jitter_buffer)
                
                # Double-check monotonic guard pasca-buffer
                current_last = self.last_applied_timestamps.get(packet.vehicle_id, 0)
                if packet.timestamp > current_last:
                    self.last_applied_timestamps[packet.vehicle_id] = packet.timestamp
                    await output_callback(packet)

            # Eviksi memori signature set secara berkala (ponytail: gunakan TTL cache untuk prod)
            if len(self.seen_signatures) > 10000:
                self.seen_signatures.clear()

            await asyncio.sleep(0.1)

# Runnable Verification Check
async def main():
    async def mock_broadcast(pkt: TelemetryPacket):
        print(f"DISPATCHED: {pkt.vehicle_id} at T={pkt.timestamp} pos=({pkt.latitude}, {pkt.longitude})")

    pipeline = GTFSIngestionPipeline(buffer_delay_sec=0.5)
    asyncio.create_task(pipeline.consumer_loop(mock_broadcast))

    t_now = int(time.time())
    
    # Simulasi kedatangan data acak (out-of-order arrival)
    pipeline.ingest("TRAIN_01", t_now + 2, -6.1754, 106.8272) # Valid
    pipeline.ingest("TRAIN_01", t_now + 1, -6.1750, 106.8270) # Tiba belakangan tapi timestamp lebih lama
    pipeline.ingest("TRAIN_01", t_now + 1, -6.1750, 106.8270) # Duplikat identik (langsung ditolak)
    pipeline.ingest("TRAIN_01", t_now + 100, -6.1750, 106.8270) # Extreme clock drift (ditolak)

    await asyncio.sleep(1.0)

if __name__ == "__main__":
    asyncio.run(main())

Analisis Trade-Off Arsitektur

Implementasi mekanisme penanganan data telemetri di atas memiliki beberapa konsekuensi operasional yang perlu dipertimbangkan:

  • Latensi Tampilan vs. Integritas Data: Penggunaan sliding jitter buffer menambahkan latensi visual buatan sebesar nilai buffer_delay_sec. Jika target sistem adalah dashboard pergerakan kereta presisi, latensi 1–2 detik adalah kompromi wajar untuk menghindari efek teleporting atau mundur. Namun, untuk sistem darurat (collision warning), buffer ini harus ditiadakan dan beralih murni ke stale drop logic.
  • Beban Memori: Menggunakan heap memory untuk menampung telemetri membutuhkan pengawasan. Jika jumlah entitas transit mencapai puluhan ribu unit dengan frekuensi pembaruan 1 Hz, pertahankan sliding window seringkas mungkin agar garbage collector tidak memicu latensi pemrosesan.

Tips Debugging Feed di Lapangan

Saat menguji integrasi langsung dengan vendor armada atau penyedia feed transit:

  1. Catat metrik stale_drop_rate dan clock_drift_reject_rate menggunakan Prometheus atau Datadog. Lonjakan mendadak pada metrik tersebut hampir selalu mengindikasikan kerusakan antena GPS atau kegagalan sinkronisasi NTP pada unit armada tertentu.
  2. Identifikasi apakah feed GTFS-RT menyediakan incremental update atau differential feed. Jika feed berupa full snapshot berkala, prioritaskan filter deduplikasi sebelum data masuk ke tahap deserialisasi protobuf penuh guna menghemat siklus CPU worker.