liminal-server 0.25.4

//! Removed lines are told before the full snapshot that follows them.

use super::{ConsoleError, ConsoleHub, PumpReport, ServerPublisher, TOPOLOGY_SUBJECT};
use crate::console::counts::{ConsoleFailure, add, failed};
use crate::console::records::{RECORD_VERSION, canonical_bytes};
use std::collections::BTreeSet;

impl ConsoleHub {
    pub(super) fn pump_topology(
        &self,
        out: &dyn ServerPublisher,
        report: &mut PumpReport,
    ) -> Result<(), ConsoleError> {
        let mut state = self.lock_topology()?;
        if !state.stale || state.subscribers == 0 {
            return Ok(());
        }
        let (mut document, fault) = state.model.document(&self.wall);
        let lines: BTreeSet<_> = document.lines.iter().map(|line| line.id.clone()).collect();
        let removed = state
            .published
            .as_ref()
            .map_or_else(Vec::new, |(_, previous)| {
                previous.difference(&lines).cloned().collect::<Vec<_>>()
            });
        let previous_rev = state.published.as_ref().map_or(0, |(rev, _)| *rev);
        let remove_count =
            u64::try_from(removed.len()).map_err(|_| ConsoleError::TopologyRevisionOverflow)?;
        // A snapshot following removal changes needs its own revision.
        state.model.ensure_revision(
            previous_rev
                .checked_add(remove_count)
                .and_then(|rev| rev.checked_add(u64::from(remove_count != 0)))
                .ok_or(ConsoleError::TopologyRevisionOverflow)?,
        );
        document.rev = state.model.rev();
        state.stale = false;
        drop(state);
        if let Some(error) = fault {
            failed(
                &self.counts.topology_clock_failures,
                1,
                ConsoleFailure::TopologyClock,
            );
            tracing::warn!(%error, "CONSOLE_TOPOLOGY_CLOCK");
        }
        for (index, id) in removed.into_iter().enumerate() {
            let rev = previous_rev
                .checked_add(
                    u64::try_from(index).map_err(|_| ConsoleError::TopologyRevisionOverflow)?,
                )
                .and_then(|rev| rev.checked_add(1))
                .ok_or(ConsoleError::TopologyRevisionOverflow)?;
            let record = serde_json::json!({"v": RECORD_VERSION, "type": "topology.change", "rev": rev, "change": {"op": "line.remove", "id": id}});
            let bytes = canonical_bytes(&record, "topology.change")?;
            if !self.send_topology(out, &bytes, report)? {
                return Ok(());
            }
        }
        let bytes = canonical_bytes(&document, "topology")?;
        if self.send_topology(out, &bytes, report)? {
            self.lock_topology()?.published = Some((document.rev, lines));
        }
        Ok(())
    }

    fn send_topology(
        &self,
        out: &dyn ServerPublisher,
        bytes: &[u8],
        report: &mut PumpReport,
    ) -> Result<bool, ConsoleError> {
        if self.send(out, TOPOLOGY_SUBJECT, bytes) {
            add(&self.counts.topology_records_published, 1);
            report.topology_records += 1;
            Ok(true)
        } else {
            report.publish_failures += 1;
            self.lock_topology()?.stale = true;
            Ok(false)
        }
    }
}