Method emit_write_event

nodedb/src/data/executor/core_loop/event_emit.rs:93–130

Emit a write event to the Event Plane. Called after a successful write (PointPut, PointDelete, PointUpdate, BatchInsert, BulkDelete, atomic KV ops, etc.). The Data Plane NEVER blocks here — if the ring buffer is full, the event is dropped and the Event Plane will detect the gap via sequence numbers and replay from WAL. Prefer [`CoreLoop::emit_put_event`] for any handler that performs a put-style

(
        &mut self,
        task: &super::super::task::ExecutionTask,
        collection: &str,
        op: crate::event::WriteOp,
        row_id: &str,
        new_value: Option<&[u8]>,
        old_

    /// determined by the operation itself (kv-atomic increment, CAS, plain
    /// delete) rather than by inspecting pre/post state.
    pub(in crate::data::executor) fn emit_write_event(
        &mut self,
        task: &super::super::task::ExecutionTask,
        collection: &str,
        op: crate::event::WriteOp,
        row_id: &str,
        new_value: Option<&[u8]>,
        old_value: Option<&[u8]>,
    ) {
        let producer = match self.event_producer.as_mut() {
            Some(p) => p,
            None => return, // Event Plane not configured.
        };

        self.event_sequence += 1;

        let (system_time_ms, valid_time_ms) =
            crate::event::bitemporal_extract::extract_stamps(new_value.or(old_value));

        let event = crate::event::WriteEvent {
            sequence: self.event_sequence,
            collection: Arc::from(collection),
            op,
            row_id: crate::event::types::RowId::new(row_id),
            lsn: self.watermark,
            tenant_id: task.request.tenant_id,
            vshard_id: task.request.vshard_id,
            source: task.request.event_source,
            new_value: new_value.map(Arc::from),
            old_value: old_value.map(Arc::from),
            system_time_ms,
            valid_time_ms,
            user_id: task.request.user_id.clone(),
            statement_digest: task.request.statement_digest.clone(),
        };

        producer.emit(event);
    }

    /// Emit a heartbeat event to advance the Event Plane's partition watermark.
    ///