Resume Tokens

Stream yang berjalan berbulan-bulan akan mengalami koneksi terputus, deploy ulang worker, atau failover — bukan pertanyaan apakah, melainkan kapan. Tanpa penanda posisi, setiap restart berarti memproses ulang seluruh aliran atau kehilangan event yang terjadi saat worker mati. Resume token menyelesaikan ini: setiap event membawa penanda posisinya sendiri di field _id, dan stream baru bisa dibuka tepat dari titik itu. Chapter ini menguji mekanismenya secara nyata — termasuk pola penyimpanan token yang dipakai worker produksi.

Token sebagai Penanda Posisi #

Field _id pada setiap change event adalah resume token: nilai opaque berupa string _data heksadesimal yang meng-encode posisi event di aliran oplog. Aturan mainnya dua: jangan pernah mem-parse atau membuat token sendiri (formatnya internal dan bisa berubah antar versi), dan gunakan token apa adanya sebagai bookmark.

Semantik resumeAfter presisi: stream baru mulai setelah event milik token itu. Semua event yang mengikuti token — yang sudah maupun yang baru terjadi — akan mengalir lagi. Kita buktikan dengan satu worker yang memproses dua event lalu menyimpan kedua tokennya:

use latihan_events
db.dropDatabase()
const cs = db.pesanan.watch()
cs.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-001", pelanggan: "Budi", total: 250000 })
db.pesanan.insertOne({ kode: "PSN-002", pelanggan: "Siti", total: 175000 })
const ev1 = cs.next()
const ev2 = cs.next()
const token1 = ev1._id
const token2 = ev2._id
cs.close()
({ diterima: [ev1.fullDocument.kode, ev2.fullDocument.kode], panjangToken: token1._data.length, tokenBerbeda: token1._data !== token2._data })

Output yang Dihasilkan:

switched to db latihan_events
{ ok: 1, dropped: 'latihan_events' }

ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedId: ObjectId('6ad3ea40acd2dcc6f8b9cfcf')
}
{
  acknowledged: true,
  insertedId: ObjectId('6ad3ea40acd2dcc6f8b9cfd0')
}

{
  diterima: [ 'PSN-001', 'PSN-002' ],
  panjangToken: 188,
  tokenBerbeda: true
}

Dua event diproses, dan setiap event membawa token berbeda — token2 selalu satu langkah di depan token1. Dua ujian berikutnya memakai kedua token itu untuk membuktikan semantik “setelah”.

Melanjutkan dari Titik Tertentu #

Pertama, kita resume dari token1 — posisi di tengah aliran. Teori kita: event PSN-002 (yang berada setelah token1) harus mengalir lagi, sementara PSN-001 tidak:

const csB = db.pesanan.watch([], { resumeAfter: token1 })
csB.disableBlockWarnings()
const evUlang = csB.next()
csB.close()
({ eventPertamaStreamBaru: evUlang.fullDocument.kode })

Output yang Dihasilkan:

ChangeStreamCursor on pesanan

{ eventPertamaStreamBaru: 'PSN-002' }

Terbukti: stream baru mengalirkan PSN-002, bukan mengulang dari PSN-001. Ini juga mengungkap sifat pengiriman change stream: at-least-once. Bila worker memproses PSN-002 lalu mati sebelum token sempat disimpan, resume dari token1 akan mengirimkan PSN-002 sekali lagi — konsumen wajib idempoten (kita bahas lengkap di Chapter Menangani Interruption).

Kedua, kita resume dari token2 — posisi terakhir yang sudah diproses. Teori kita: hanya event baru yang muncul:

const csC = db.pesanan.watch([], { resumeAfter: token2 })
csC.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-003", pelanggan: "Dewi", total: 310000 })
const ev3 = csC.next()
db.checkpoint.insertOne({ _id: "worker-notifikasi", tokenTerakhir: ev3._id, diperbarui: new Date() })
csC.close()
({ eventBaru: ev3.fullDocument.kode })

Output yang Dihasilkan:

ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedId: ObjectId('6ad3ea41acd2dcc6f8b9cfd1')
}

{ acknowledged: true, insertedId: 'worker-notifikasi' }

{ eventBaru: 'PSN-003' }

Stream ketiga tidak mengulang PSN-001 maupun PSN-002 — next() langsung memberikan PSN-003, event pertama setelah token2. Inilah pasangan aturan yang membuat resume aman: event sebelum token tidak pernah dikirim ulang, event setelah token tidak pernah terlewat.

Menyimpan Token di Aplikasi #

Token baru berguna bila bertahan melebihi umur stream. Pola produksinya: setelah satu event selesai diproses (side effect sukses), simpan tokennya ke penyimpanan tahan gangguan — collection checkpoint, Redis, atau file. Saat worker menyala kembali, token itu menjadi titik lanjut. Blok terakhir mensimulasikan restart penuh: token dibaca kembali dari collection checkpoint:

const tersimpan = db.checkpoint.findOne({ _id: "worker-notifikasi" })
const csD = db.pesanan.watch([], { resumeAfter: tersimpan.tokenTerakhir })
csD.disableBlockWarnings()
db.pesanan.insertOne({ kode: "PSN-004", pelanggan: "Eka", total: 98000 })
const ev4 = csD.next()
csD.close()
({ tokenRoundtripSama: tersimpan.tokenTerakhir._data === ev3._id._data, eventSetelahRestart: ev4.fullDocument.kode })

Output yang Dihasilkan:

ChangeStreamCursor on pesanan
{
  acknowledged: true,
  insertedId: ObjectId('6ad3ea41acd2dcc6f8b9cfd2')
}

{ tokenRoundtripSama: true, eventSetelahRestart: 'PSN-004' }

Token hasil bolak-balik lewat collection tetap valid — resumeAfter menerimanya langsung, dan stream hanya mengalirkan PSN-004. Urutan penulisan di sini penting: simpan token setelah side effect selesai, bukan sebelum; kalau terbalik, crash di tengah berarti event terlewat.

Tiga Opsi Titik Mulai #

resumeAfter bukan satu-satunya cara menentukan titik mulai. Tiga opsi watch() saling melengkapi:

Perbedaan resumeAfter vs startAfter hanya muncul setelah event invalidate — kita buktikan keduanya dengan error dan event nyata di Chapter Menangani Interruption. startAtOperationTime menerima Timestamp seperti clusterTime event; presisinya lebih rendah dari token (banyak event bisa berbagi satu detik clock), jadi token tetap pilihan utama untuk checkpoint periodik.

Rangkuman #

• _id setiap event adalah resume token — opaque, jangan di-parse, jangan dibuat sendiri.
• resumeAfter: token memulai stream setelah event milik token: event lama tidak berulang, event setelah token tidak terlewat; pengiriman bersifat at-least-once.
• Simpan token ke penyimpanan tahan gangguan setelah side effect selesai diproses; baca kembali saat worker menyala untuk melanjutkan tanpa gap.
• Token berumur selama jendela oplog; di luar itu resume gagal dan butuh sinkronisasi ulang penuh.
• startAfter untuk lanjut setelah invalidate; startAtOperationTime untuk titik mulai berbasis waktu.