Change stream yang berjalan berbulan-bulan hampir pasti mengalami putus: koneksi jaringan berfluktuasi, primary berganti saat failover, aplikasi di-deploy ulang, atau collection yang ditonton dihapus. Desain yang sehat menjadikan semua itu kejadian rutin, bukan insiden: stream dideteksi putusnya, disambung ulang dari resume token, dan efek samping konsumen tetap terjadi tepat satu kali. Chapter ini menguji ketiga bekal itu dengan event dan error nyata.

Kapan Stream Terputus #

Cursor change stream tertutup karena empat penyebab:

  • Gangguan jaringan antara aplikasi dan cluster — error muncul saat getMore berikutnya.
  • Failover — election primary baru memutus cursor yang merujuk node lama.
  • Restart aplikasi saat deploy — putus yang direncanakan; resume token membuatnya mulus.
  • Event invalidate — collection di-drop atau di-rename; cursor ditutup permanen dan tidak bisa lanjut di stream yang sama.

Tiga penyebab pertama bersifat sementara; yang keempat final. Kita buktikan dulu yang keempat, karena perilakunya paling mengejutkan:

js 
use latihan_events
db.dropDatabase()
db.pesanan.insertOne({ kode: "PSN-001", pelanggan: "Budi", total: 250000 })
const cs = db.pesanan.watch()
cs.disableBlockWarnings()
db.pesanan.drop()
const evDrop = cs.next()
const evInv = cs.next()
({ jenisEvent: [evDrop.operationType, evInv.operationType], isiInvalidate: evInv })
try { cs.next() } catch (err) { print("next() setelah invalidate -> " + err.message) }
cs.close()

Output yang Dihasilkan:

text 
switched to db latihan_events
{ ok: 1, dropped: 'latihan_events' }
{
  acknowledged: true,
  insertedId: ObjectId('6ad3eafb6038667d0426c0b5')
}

ChangeStreamCursor on pesanan
true

{
  jenisEvent: [ 'drop', 'invalidate' ],
  isiInvalidate: {
    _id: {
      _data: '826AD3EAFC000000012B042C0100296E5A100427697A43BCE74C95B1A1C41E42F1EB5F463C6F7065726174696F6E54797065003C64726F70000004'
    },
    operationType: 'invalidate',
    clusterTime: Timestamp({ t: 1792273148, i: 1 }),
    wallTime: ISODate('2026-10-17T21:39:08.771Z')
  }
}
next() setelah invalidate -> ChangeStream is closed

Meng-drop collection menghasilkan dua event berurutan: drop lalu invalidate. Event invalidate berukuran paling kecil dari semua jenis event — hanya _id, operationType, clusterTime, dan wallTime, tanpa ns maupun documentKey — karena yang diumumkan adalah berakhirnya stream itu sendiri, bukan perubahan data. Panggilan next() setelahnya gagal: cursor sudah ditutup server.

resumeAfter vs startAfter Setelah Invalidate #

Pertanyaan berikutnya: bisakah stream dilanjutkan setelah invalidate? Jawabannya bergantung opsi yang dipakai. Kita coba keduanya dengan token dari event invalidate di atas, pada collection yang dibuat ulang:

js 
db.createCollection("pesanan")
db.pesanan.insertOne({ kode: "PSN-100", pelanggan: "Siti", total: 120000 })
const csA = db.pesanan.watch([], { resumeAfter: evInv._id })
csA.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-101", pelanggan: "Dewi", total: 210000 })
try { csA.next() } catch (err) { print("resumeAfter token invalidate -> " + err.message) }
csA.close()
const csB = db.pesanan.watch([], { startAfter: evInv._id })
csB.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-102", pelanggan: "Eka", total: 64000 })
({ startAfterEventPertama: csB.next().fullDocument.kode })
csB.close()

Output yang Dihasilkan:

text 
{ ok: 1 }
{
  acknowledged: true,
  insertedId: ObjectId('6ad3eafc6038667d0426c0b6')
}

ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedId: ObjectId('6ad3eafc6038667d0426c0b7')
}
resumeAfter token invalidate -> Attempting to resume a change stream using 'resumeAfter' is not allowed from an invalidate notification

ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedId: ObjectId('6ad3eafc6038667d0426c0b8')
}
{ startAfterEventPertama: 'PSN-100' }

resumeAfter ditolak keras oleh server: token dari event invalidate tidak boleh dipakai untuk resume. startAfter menerima token yang sama dan stream baru langsung hidup — bahkan mengalirkan event yang terjadi setelah invalidate pada collection yang dibuat ulang, mulai dari insert PSN-100. Untuk drop yang disengaja (migrasi schema), worker biasanya memang berhenti; startAfter relevan saat invalidate datang dari rename atau drop yang tidak Anda antisipasi.

Pola Reconnect dengan Token #

Untuk tiga penyebab sementara, solusinya satu kata: ulang dari token terakhir. Kerangka worker produksinya berbentuk loop — di sini kita jalankan dua iterasi agar outputnya terbaca, dengan setiap iterasi mensimulasikan satu “hidup” worker:

js 
let tokenTerakhir = null
const diproses = []
for (let sesi = 1; sesi <= 2; sesi++) {
  try {
    const opsi = tokenTerakhir === null ? {} : { resumeAfter: tokenTerakhir }
    const cs = db.pesanan.watch([], opsi)
    cs.disableBlockWarnings()
    db.pesanan.insertOne({ kode: "PSN-11" + sesi, pelanggan: "Budi", total: 100000 + sesi })
    const ev = cs.next()
    tokenTerakhir = ev._id
    diproses.push(ev.fullDocument.kode)
    cs.close()
  } catch (err) {
    print("Gangguan (" + err.codeName + ") - stream dibuka ulang")
  }
}
({ diproses: diproses, tokenCheckpointAda: tokenTerakhir !== null })

Output yang Dihasilkan:

text 
{ diproses: [ 'PSN-111', 'PSN-112' ], tokenCheckpointAda: true }

Sesi pertama membuka stream baru (tanpa token), memproses PSN-111, lalu “mati”. Sesi kedua membuka stream dengan resumeAfter token PSN-111 — PSN-111 tidak terkirim ulang, dan PSN-112 diproses normal. Di produksi, for diganti while (true) dan catch menambah jeda exponential backoff agar server yang sedang sibuk failover tidak ditimbani permintaan baru.

Kapan boleh mengulang? Spesifikasi change streams MongoDB membagi error jadi dua golongan. Driver resmi otomatis mencoba menyambung ulang satu kali untuk error resumable — gangguan jaringan, atau error server berlabel ResumableChangeStreamError seperti CursorNotFound. Contoh nyatanya saat failover: getMore mengembalikan InterruptedDueToReplStateChange (kode 11602), dan driver menyambungkan ulang stream otomatis dari token terakhir. Error lain — termasuk token yang sudah di luar jendela oplog — dilempar ke kode kita; mengulanginya tidak akan menyembuhkan apa pun, jadi loop harus menghentikan worker dan memicu pemulihan penuh.

Dua opsi cursor melengkapi ketahanan ini. maxAwaitTimeMS membatasi berapa lama server menahan satu getMore sebelum mengembalikan batch kosong; batch kosong itu membawa postBatchResumeToken — sinyal periodak bahwa stream tetap hidup, sekaligus token cadangan yang aman disimpan ketika event jarang mengalir.

Konsumen Idempoten #

Resume menjamin at-least-once: bila worker memproses event lalu berhenti sebelum tokennya tersimpan, event itu akan terkirim lagi setelah sambung ulang. Efek samping (kirim notifikasi, tulis ke sistem eksternal) harus dilindungi dari pemrosesan ganda — pola paling sederhana: catat kunci documentKey + clusterTime yang sudah ditangani:

js 
const sudahDiproses = new Set()
const efekSamping = []
const serapEvent = (ev) => {
  const kunci = String(ev.documentKey._id) + "@" + String(ev.clusterTime)
  if (sudahDiproses.has(kunci)) return "duplikat dilewati"
  sudahDiproses.add(kunci)
  efekSamping.push(ev.fullDocument.kode)
  return "diproses"
}

// sesi 1: event diproses, tetapi worker berhenti SEBELUM token disimpan
const cs1 = db.pesanan.watch()
cs1.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-120", pelanggan: "Siti", total: 47000 })
print("sesi 1: " + serapEvent(cs1.next()))
cs1.close()

// sesi 2: resume dari token lama -> event yang sama terkirim ulang
const cs2 = db.pesanan.watch([], { resumeAfter: tokenTerakhir })
cs2.disableBlockWarnings()
print("sesi 2: " + serapEvent(cs2.next()))
cs2.close()
({ efekSamping: efekSamping, eventUnik: sudahDiproses.size })

Output yang Dihasilkan:

text 
ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedId: ObjectId('6ad3eaff6038667d0426c0bb')
}
sesi 1: diproses

ChangeStreamCursor on pesanan
sesi 2: duplikat dilewati

{ efekSamping: [ 'PSN-120' ], eventUnik: 1 }

Sesi 1 memproses PSN-120, lalu worker “mati” dengan tokenTerakhir masih tertinggal di PSN-112. Sesi 2 resume dari token itu — PSN-120 terkirim lagi — tetapi serapEvent mengenali kunci _id + clusterTime yang sama dan melewatinya. Hasil akhirnya: satu efek samping untuk satu event, terlepas dari berapa kali pengiriman. Di sistem produksi, Set ini menjadi tabel event_diproses dengan TTL, atau idempotency key di sistem tujuan.

Rangkuman #

  • Stream putus karena jaringan, failover, restart, atau invalidate (drop/rename collection); tiga pertama sementara, invalidate final.
  • Drop collection menghasilkan event drop lalu invalidate; setelah invalidate, cursor tertutup dan next() melempar error.
  • resumeAfter menolak token invalidate; startAfter menerimanya dan mengalirkan event setelah invalidate, termasuk dari collection yang dibuat ulang.
  • Worker produksi = loop try/catch dengan resumeAfter token terakhir + backoff; driver otomatis menyambung ulang sekali untuk error resumable (berlabel ResumableChangeStreamError).
  • maxAwaitTimeMS mengatur jeda batch kosong; postBatchResumeToken pada batch kosong adalah heartbeat dan token cadangan.
  • Pengiriman bersifat at-least-once: dedup dengan kunci documentKey + clusterTime menjaga efek samping tetap satu kali.