Sinkronisasi data antar sistem terdistribusi sering kali diawali dengan pendekatan sederhana: full snapshot sync. Setiap interval tertentu, klien memanggil endpoint seperti GET /api/v1/resources untuk mengunduh seluruh dataset. Pola ini runtuh saat volume data menembus puluhan ribu record: latensi melonjak, utilisasi bandwidth boros, dan beban I/O database meningkat drastis akibat serialization overhead.

Solusinya adalah incremental sync, di mana hanya perubahan (delta) yang ditransmisikan. Namun, beralih ke delta update memicu kompleksitas baru: state gap akibat jaringan putus, race condition dari webhook retry, dan inkonsistensi data ketika event tiba di luar urutan (out-of-order delivery).

Paradigma Incrementality: Mengadaptasi Dependency Graph & Memoization

Konsep incremental sync memiliki kesamaan matematis dengan compiler modern seperti Typst atau Rust Analyzer. Typst tidak mem-parsing ulang seluruh dokumen saat satu karakter berubah. Typst memetakan dokumen ke dalam Directed Acyclic Graph (DAG), membandingkan hash/versi pada node input, me-memoize komputasi node yang statis, dan hanya mengeksekusi ulang node yang bertanda dirty.

Pada arsitektur API, prinsip ini diterapkan melalui model berikut:

  • Deterministic Change Tracking (DAG/Dirty State): Database sumber menandai entitas yang berubah menggunakan sequence id yang monotonik naik (monotonic sequence) atau transaction commit timestamp.
  • Memoized Read Model: Klien menyimpan state lokal terkahir yang valid. Endpoint sync mengevaluasi selisih antara pointer klien (cursor) dan state terbaru di database sumber.
  • Atomic State Invalidation: Perubahan downstream hanya dipicu pada level atribut yang termodifikasi, bukan rekalkulasi seluruh relasi entitas.

Pitfall Transisi: Dari Snapshot ke Delta Updates

Mengganti snapshot dengan stream delta membuka celah desinkronisasi jika kontrak data tidak dirancang secara defensif:

  1. State Gap (Missing Events): Jika network timeout memutus HTTP stream atau webhook gagal terkirim melampaui retry policy, klien kehilangan satu atau lebih event delta. Tanpa pendeteksi gap, klien menerapkan perubahan di atas basis state yang salah.
  2. Race Condition Akibat Retries: Mekanisme retry pada webhook tanpa sorting key menyebabkan update lama (misal: status = PENDING) menimpa update baru (status = SUCCESS) yang tiba lebih dulu.
  3. Broken Idempotency pada Delta Relatif: Payload seperti {"balance_delta": -5000} tidak idempotent jika diulang saat network glitch. Delta harus diformulasikan secara deterministik atau mereferensikan sequence version spesifik.

Desain Kontrak API: Cursor vs Version Vector & Skema Deterministik

Hindari penggunaan updated_at standar berbasis wall-clock timestamp sebagai cursor. Clock skew antar server dan duplikasi timestamp pada transaksi konkuren akan melompati record. Gunakan Monotonic Sequence ID atau Logical Versioning per entitas.

Contoh Kontrak Delta Payload

{
  "sync_meta": {
    "cursor": "seq_1084920",
    "has_more": false,
    "server_time": "2024-10-24T12:00:00Z"
  },
  "changes": [
    {
      "entity_id": "usr_88291",
      "version": 4,
      "previous_version": 3,
      "operation": "UPSERT",
      "payload": {
        "email": "[email protected]",
        "tier": "premium"
      }
    },
    {
      "entity_id": "usr_10283",
      "version": 12,
      "previous_version": 11,
      "operation": "DELETE",
      "payload": null
    }
  ]
}

Menyertakan version dan previous_version secara eksplisit membuat payload bersifat self-verifying. Klien dapat langsung memvalidasi kontinuitas delta tanpa bergantung pada metadata eksternal.

Implementasi Delta Validator & Idempotency Guard

Berikut implementasi minimalis consumer sync menggunakan Python. Kode ini menangani deduplication, out-of-order buffering, dan deteksi gap tanpa dependensi framework tambahan.

import sqlite3
from typing import Dict, Any, Optional

class DeltaSyncConsumer:
    def __init__(self, db_conn: sqlite3.Connection):
        self.db = db_conn
        self._init_schema()

    def _init_schema(self):
        with self.db:
            self.db.execute("""
                CREATE TABLE IF NOT EXISTS entities (
                    id TEXT PRIMARY KEY,
                    version INTEGER NOT NULL,
                    data TEXT NOT NULL
                )
            """)
            self.db.execute("""
                CREATE TABLE IF NOT EXISTS delta_buffer (
                    entity_id TEXT,
                    version INTEGER,
                    prev_version INTEGER,
                    payload TEXT,
                    PRIMARY KEY (entity_id, version)
                )
            """)

    def apply_delta(self, change: Dict[str, Any]) -> str:
        entity_id = change["entity_id"]
        incoming_version = change["version"]
        prev_version = change["previous_version"]
        raw_payload = str(change.get("payload"))

        with self.db:
            row = self.db.execute(
                "SELECT version FROM entities WHERE id = ?", (entity_id,)
            ).fetchone()
            current_version = row[0] if row else 0

            # 1. Idempotency Check: abaikan update lama atau duplikat
            if incoming_version <= current_version:
                return "IGNORED_DUPLICATE_OR_STALE"

            # 2. Strict Continuity Check: prev_version harus cocok persis
            if prev_version != current_version:
                # State gap terdeteksi; simpan ke buffer penampung
                # ponytail: in-memory/sqlite buffer; swap to Redis sorted set for distributed consumers.
                self.db.execute(
                    "INSERT OR REPLACE INTO delta_buffer VALUES (?, ?, ?, ?)",
                    (entity_id, incoming_version, prev_version, raw_payload)
                )
                return "GAP_DETECTED_BUFFERED"

            # 3. Terapkan state update secara atomic
            self.db.execute("""
                INSERT INTO entities (id, version, data)
                VALUES (?, ?, ?)
                ON CONFLICT(id) DO UPDATE SET
                    version = excluded.version,
                    data = excluded.data
            """, (entity_id, incoming_version, raw_payload))

            # 4. Drain buffer jika ada pending delta yang berurutan
            self._drain_buffer(entity_id, incoming_version)
            return "APPLIED"

    def _drain_buffer(self, entity_id: str, current_version: int):
        next_row = self.db.execute(
            "SELECT version, payload FROM delta_buffer WHERE entity_id = ? AND prev_version = ?",
            (entity_id, current_version)
        ).fetchone()

        if next_row:
            v, p = next_row
            self.db.execute(
                "UPDATE entities SET version = ?, data = ? WHERE id = ?",
                (v, p, entity_id)
            )
            self.db.execute(
                "DELETE FROM delta_buffer WHERE entity_id = ? AND version = ?",
                (entity_id, v)
            )
            self._drain_buffer(entity_id, v)

Skipped: lock distributed multi-worker. Tambahkan ketika konsumsi delta dijalankan di lebih dari satu instance worker paralel.

Mekanisme Auto-Reconciliation saat Terdeteksi Desync

Ketika klien mendeteksi status GAP_DETECTED_BUFFERED, buffer lokal tidak boleh menahan event tanpa batas waktu. Jika missing link tidak tiba dalam jendela waktu tertentu (misal: 30 detik), sistem harus menginisiasi Automatic Reconciliation Cycle:

  1. Selective Range Fetch: Klien mengeksekusi request sinkronisasi manual: GET /api/v1/entities/{id}/deltas?from_version={current_version}&to_version={target_version}.
  2. Fallback ke Snapshot Per-Entitas: Jika log delta pada server telah dibersihkan (compacted/purged), endpoint delta harus mengembalikan response 410 Gone. Klien kemudian beralih mengambil single-entity snapshot via GET /api/v1/entities/{id} untuk mereset base version.
  3. Poison Pill Quarantine: Jika entitas berulang kali gagal divalidasi setelah 3 kali siklus rekonsiliasi, tandai entitas tersebut di local store sebagai DESYNC_DIRTY dan trigger background alert ke sistem observabilitas tanpa menghentikan sinkronisasi record lainnya.
Delta sync yang tangguh tidak mengasumsikan jaringan andal. Ketahanan sistem ditentukan oleh kemampuan kontrak data mendeteksi anomali secara lokal dan melakukan self-healing tanpa intervensi manual.