Emit a heartbeat event to advance the Event Plane's partition watermark. Called when no user writes occur for >1 second. The heartbeat carries the current watermark LSN so the Event Plane can advance its partition watermark without waiting for user writes.
(&mut self)
| 135 | /// the current watermark LSN so the Event Plane can advance its partition |
| 136 | /// watermark without waiting for user writes. |
| 137 | pub fn emit_heartbeat(&mut self) { |
| 138 | let producer = match self.event_producer.as_mut() { |
| 139 | Some(p) => p, |
| 140 | None => return, |
| 141 | }; |
| 142 | |
| 143 | self.event_sequence += 1; |
| 144 | |
| 145 | let event = crate::event::WriteEvent { |
| 146 | sequence: self.event_sequence, |
| 147 | collection: Arc::from("_heartbeat"), |
| 148 | op: crate::event::WriteOp::Heartbeat, |
| 149 | row_id: crate::event::types::RowId::new(""), |
| 150 | // watermark = last committed LSN. Correct for heartbeats: uncommitted |
| 151 | // writes should NOT advance the Event Plane's watermark. |
| 152 | lsn: self.watermark, |
| 153 | // Default tenant; vshard derived from core_id for partition routing. |
| 154 | tenant_id: crate::types::TenantId::new(0), |
| 155 | vshard_id: crate::types::VShardId::new( |
| 156 | (self.core_id % crate::types::VShardId::COUNT as usize) as u32, |
| 157 | ), |
| 158 | source: crate::event::EventSource::User, |
| 159 | new_value: None, |
| 160 | old_value: None, |
| 161 | system_time_ms: None, |
| 162 | valid_time_ms: None, |
| 163 | user_id: None, |
| 164 | statement_digest: None, |
| 165 | }; |
| 166 | |
| 167 | producer.emit(event); |
| 168 | } |
| 169 | } |