Method send_append_entries_to_observer

nodedb-raft/src/node/internal.rs:227–316

Send `AppendEntries` to an observer peer. If the observer's advisory send queue is full (backpressure threshold reached), the send is skipped. The observer will fall behind and recover via snapshot when it reconnects. Source commits are never delayed.

(&mut self, observer: u64)

    /// reached), the send is skipped. The observer will fall behind and recover
    /// via snapshot when it reconnects. Source commits are never delayed.
    pub(super) fn send_append_entries_to_observer(&mut self, observer: u64) {
        let can_receive = match &self.leader_state {
            Some(ls) => ls.observer_can_receive(observer),
            None => return,
        };
        if !can_receive {
            debug!(
                node = self.config.node_id,
                group = self.config.group_id,
                observer,
                "observer send queue full; skipping (advisory backpressure)"
            );
            return;
        }

        let leader = match &self.leader_state {
            Some(ls) => ls,
            None => return,
        };

        let obs_state = match leader
            .observer_states
            .iter()
            .find(|(id, _)| *id == observer)
        {
            Some((_, s)) => s.clone(),
            None => return,
        };

        let next_index = obs_state.next_index;
        let prev_log_index = next_index.saturating_sub(1);

        let prev_log_term = match self.log.term_at(prev_log_index) {
            Some(term) => term,
            None => {
                debug!(
                    node = self.config.node_id,
                    group = self.config.group_id,
                    observer,
                    next_index,
                    snapshot_index = self.log.snapshot_index(),
                    "observer needs snapshot (log compacted)"
                );
                self.ready.snapshots_needed.push(observer);
                return;
            }
        };

        let entries = if next_index <= self.log.last_index() {
            match self.log.entries_range(next_index, self.log.last_index()) {
                Ok(slice) => slice.to_vec(),
                Err(crate::error::RaftError::LogCompacted { .. }) => {
                    debug!(
                        node = self.config.node_id,
                        group = self.config.group_id,
                        observer,
                        next_index,
                        "observer needs snapshot (entries compacted)"