ozzy_replication/driver/

exit_view.rs

//! Quorum-supported timeout suspicion, independent of durable election evidence.

use super::{Action, DriverError, ReplicaDriver, Role};
use crate::wire::Control;
use crate::{Scope, StartViewChange, VoterIndex, VoterSet};
use ozzy_proto::NodeId;
use std::time::Duration;

impl ReplicaDriver {
    pub(super) fn receive_exit(
        &mut self,
        voter: VoterIndex,
        scope: Scope,
    ) -> Result<(), DriverError> {
        self.require_healthy()?;
        // Neither a distant wish nor a duplicate from another view is evidence
        // that anyone has entered that view. START_VIEW_CHANGE supplies that fence.
        if scope.view < self.scope().view {
            // This voter left that view, possibly on the sender's own request
            // plus a local timeout, before its suspicion went out. Without
            // support the sender cannot leave until this voter's durable
            // promise lets START_VIEW_CHANGE reach it. After that promise,
            // election traffic carries the newer view while it lasts.
            //
            // A peer that also left the view cannot tell this answer from a
            // request. Answer only the sender, and each sender at most once
            // per retransmission interval, so two such voters fall silent.
            if self.now < self.exit_echo_at[voter] {
                return Ok(());
            }
            if self.promise_pending() {
                self.exit_echo_at[voter] = self.now.saturating_add(self.timing.retransmit);
                let mut asked = match self.exit_echo {
                    Some((old, asked)) if old == scope => asked,
                    _ => VoterSet::EMPTY,
                };
                asked.insert(voter);
                self.exit_echo = Some((scope, asked));
            } else if matches!(self.role, Role::Normal(_) | Role::Installing(_)) {
                // Election traffic stops once a view is normal, and normal
                // traffic carries the view only from its primary. A sender
                // that cannot hear that primary would never learn of the
                // view, and with the primary gone the two survivors could
                // never form another. Answer with this view's durable fence
                // instead, which brings the sender into it.
                self.exit_echo_at[voter] = self.now.saturating_add(self.timing.retransmit);
                self.fence_answers.insert(voter);
            }
            return Ok(());
        }
        if scope.view > self.scope().view {
            // A request to leave a newer view says nothing about this one,
            // but it shows this view is stale. An idle primary has no timeout
            // of its own, and only the newer view's primary would tell it of
            // that view. With that primary gone, the sender could never form
            // a view with it. Ask the sender to leave this view instead: its
            // answer is the newer view's fence, which brings this voter in.
            if matches!(self.role, Role::Normal(_)) && self.now >= self.exit_echo_at[voter] {
                self.exit_echo_at[voter] = self.now.saturating_add(self.timing.retransmit);
                self.stale_answers.insert(voter);
            }
            return Ok(());
        }
        if scope != self.scope() {
            return Ok(());
        }
        if let Some(normal) = self.normal() {
            let snapshot = normal.snapshot();
            if !self.normal_timed_out(&snapshot) {
                self.withdraw_exit();
            }
        }
        self.exit_heard_at[voter] = self.now;
        self.exit_votes.insert(voter);
        self.advance_if_exit_quorum()
    }

    /// `from` asked for recovery as a nonvoting copy. When it is the primary of the
    /// view this replica elects, it cannot install that view, so suspect the
    /// view now instead of at the election deadline. Leaving the view still
    /// needs a second voter's suspicion.
    pub fn observe_recovering(&mut self, from: NodeId, now: Duration) -> Result<(), DriverError> {
        self.observe_time(now)?;
        self.require_healthy()?;
        let voter = self.configuration.voter_index(from)?;
        if voter != self.local
            && voter == VoterIndex::primary(self.scope().view)
            && matches!(self.role, Role::ViewChanging(_))
        {
            self.request_exit()?;
        }
        Ok(())
    }

    pub(super) fn request_exit(&mut self) -> Result<(), DriverError> {
        self.exit_retry_at.get_or_insert(self.now);
        self.exit_votes.insert(self.local);
        self.advance_if_exit_quorum()
    }

    fn advance_if_exit_quorum(&mut self) -> Result<(), DriverError> {
        // A suspicious peer repeats its request every retransmission
        // interval. One that recovered goes quiet, so its request lapses
        // instead of helping a later timeout end a healthy view.
        for voter in self.local.others() {
            let heard = self.exit_heard_at[voter];
            if self.now >= heard.saturating_add(self.timing.primary_timeout) {
                self.exit_votes.remove(voter);
            }
        }
        if self.exit_votes.len() >= 2 {
            if self
                .normal()
                .is_none_or(|normal| !normal.snapshot().ready_for_appends)
            {
                self.election_delay = self
                    .election_delay
                    .saturating_mul(2)
                    .min(self.timing.max_election_timeout);
            }
            self.next_view()?;
        }
        Ok(())
    }

    pub(super) fn withdraw_exit(&mut self) {
        self.exit_votes.remove(self.local);
        self.exit_retry_at = None;
    }

    pub(super) fn promise_pending(&self) -> bool {
        matches!(
            &self.role,
            Role::ViewChanging(changing) if changing.promised_view() != changing.scope().view
        )
    }

    /// Answer one sender's request to leave another view per call: support
    /// for an older view's exit, that view's fence, or this stale view's exit.
    /// The sender retransmits, so a lost answer is repeated after that
    /// interval.
    pub(super) fn exit_echo_action(&mut self) -> Option<Action> {
        if let Some((scope, mut asked)) = self.exit_echo.take()
            && let Some(voter) = asked.pop_first()
        {
            if !asked.is_empty() {
                self.exit_echo = Some((scope, asked));
            }
            return Some(Action::Send {
                to: self.configuration.voter(voter),
                message: Control::ExitView(scope),
            });
        }
        if let Some(voter) = self.fence_answers.pop_first() {
            return Some(Action::Send {
                to: self.configuration.voter(voter),
                message: Control::StartViewChange(StartViewChange {
                    scope: self.scope(),
                }),
            });
        }
        let voter = self.stale_answers.pop_first()?;
        Some(Action::Send {
            to: self.configuration.voter(voter),
            message: Control::ExitView(self.scope()),
        })
    }

    pub(super) fn exit_action(&mut self) -> Option<Action> {
        if self.now < self.exit_retry_at? {
            return None;
        }
        self.exit_retry_at = Some(self.now.saturating_add(self.timing.retransmit));
        Some(Action::Broadcast(Control::ExitView(self.scope())))
    }
}