haematite-cli 0.21.3

//! `head.follow`: a branch's head pushed to one connection as it moves.
//!
//! Every branch head becomes visible at one point, the catalogue's publish or
//! removal, and each of those wakes every follower (storage/catalogue.rs). A
//! woken follower reads its branch's head again and sends a frame only when
//! the version differs from the last one it sent, so a publish of another
//! branch wakes it and sends nothing. Nothing asks on a timer: the one bounded
//! wait is the presented pass's own expiry, a deadline, after which the follow
//! ends `pass_expired`. A branch no longer published ends it `branch_unknown`.
//!
//! Socket readiness wakes a cancelled follower independently of branch changes.

use std::io::{self, Read, Write};
use std::os::fd::AsRawFd;
use std::os::unix::net::UnixStream;
use std::sync::atomic::{AtomicBool, Ordering};
use std::task::Poll;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use serde_json::{Value, json};

use super::protocol::Version;
use super::storage::{Catalogue, Failure};

/// Follow `branch` in `catalogue`, giving each head frame to `send`.
///
/// The current head goes at once, then one each time its published version
/// changes. `expires` is the presented pass's expiry in seconds since the
/// epoch, read against `clock`; `None` follows with no expiry.
///
/// Answers the refusal that ends the follow, for the caller to send as the
/// last frame: `pass_expired`, `branch_unknown`, or the catalogue's own.
///
/// # Errors
///
/// The first error `send` returns: the client went away, and the follow
/// ends with nothing more to send.
pub fn follow(
    catalogue: &Catalogue,
    branch: &str,
    expires: Option<u64>,
    clock: impl Fn() -> SystemTime,
    send: impl FnMut(&Value) -> io::Result<()>,
) -> io::Result<Failure> {
    follow_cancellable(
        catalogue,
        branch,
        expires,
        clock,
        &AtomicBool::new(false),
        send,
    )
}

pub(in crate::serve) fn follow_cancellable(
    catalogue: &Catalogue,
    branch: &str,
    expires: Option<u64>,
    clock: impl Fn() -> SystemTime,
    cancelled: &AtomicBool,
    mut send: impl FnMut(&Value) -> io::Result<()>,
) -> io::Result<Failure> {
    // The count is read before the head, so a change made between the two
    // wakes the first wait at once and is read then.
    let mut seen = catalogue.changes();
    let mut last = match catalogue.head(branch) {
        Ok(Some(version)) => version,
        Ok(None) => return Ok(unknown(branch)),
        Err(failure) => return Ok(failure),
    };
    if let Err(failure) = remaining(expires, clock()) {
        return Ok(failure);
    }
    send(&head(branch, last))?;
    loop {
        connected(cancelled)?;
        let left = match remaining(expires, clock()) {
            Ok(left) => left,
            Err(failure) => return Ok(failure),
        };
        seen = catalogue.changed_past_or_closed(seen, left, cancelled);
        connected(cancelled)?;
        // Checked again on waking, so no head is sent past the expiry.
        if let Err(failure) = remaining(expires, clock()) {
            return Ok(failure);
        }
        match catalogue.head(branch) {
            Ok(Some(version)) if version != last => {
                send(&head(branch, version))?;
                last = version;
            }
            Ok(Some(_)) => {}
            Ok(None) => return Ok(unknown(branch)),
            Err(failure) => return Ok(failure),
        }
    }
}

fn connected(cancelled: &AtomicBool) -> io::Result<()> {
    if cancelled.load(Ordering::Acquire) {
        Err(io::Error::new(
            io::ErrorKind::ConnectionAborted,
            "follow client closed",
        ))
    } else {
        Ok(())
    }
}

/// One socket-readiness waiter, stopped by its own socket signal.
/// Application socket flags stay unchanged. Only the private stop socket
/// is read, nonblocking, to distinguish a stop from spurious readiness.
pub(crate) struct Monitor {
    stop: UnixStream,
    worker: Option<std::thread::JoinHandle<()>>,
}

impl Monitor {
    pub(crate) fn start(
        peer: impl AsRawFd + Send + 'static,
        closed: impl FnOnce() + Send + 'static,
    ) -> io::Result<Self> {
        Self::start_observed(peer, closed, |_| Ok(()))
    }

    fn start_observed(
        peer: impl AsRawFd + Send + 'static,
        closed: impl FnOnce() + Send + 'static,
        observed: impl Fn(bool) -> io::Result<()> + Send + 'static,
    ) -> io::Result<Self> {
        use tokio::io::unix::AsyncFd;
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()?;
        let (stop, stopped) = UnixStream::pair()?;
        stopped.set_nonblocking(true)?;
        let (peer, stopped) = {
            let _entered = runtime.enter();
            (AsyncFd::new(peer)?, AsyncFd::new(stopped)?)
        };
        let worker = std::thread::Builder::new()
            .name("haematite-follow-close".to_owned())
            .spawn(move || {
                let result: io::Result<bool> = runtime.block_on(std::future::poll_fn(|context| {
                    match stop_ready(&stopped, context) {
                        Poll::Ready(Ok(())) => return Poll::Ready(Ok(false)),
                        Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
                        Poll::Pending => {}
                    }
                    loop {
                        match peer.poll_read_ready(context) {
                            Poll::Ready(Ok(mut ready)) => {
                                let hangup =
                                    ready.ready().is_read_closed() || ready.ready().is_error();
                                observed(hangup)?;
                                if hangup {
                                    return Poll::Ready(Ok(true));
                                }
                                // Leave queued frames for the connection after its follow ends.
                                // Poll again after clearing to register the next close wake.
                                ready.clear_ready();
                            }
                            Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
                            Poll::Pending => return Poll::Pending,
                        }
                    }
                }));
                match result {
                    Ok(false) => {}
                    Ok(true) => closed(),
                    Err(error) => {
                        eprintln!("haematite follow_monitor_failed: {error}");
                        closed();
                    }
                }
            })?;
        Ok(Self {
            stop,
            worker: Some(worker),
        })
    }
}

fn stop_ready(
    signal: &tokio::io::unix::AsyncFd<UnixStream>,
    context: &mut std::task::Context<'_>,
) -> Poll<io::Result<()>> {
    loop {
        let mut ready = std::task::ready!(signal.poll_read_ready(context))?;
        let mut input = signal.get_ref();
        match input.read(&mut [0]) {
            Ok(_) => return Poll::Ready(Ok(())),
            Err(error) if error.kind() == io::ErrorKind::WouldBlock => ready.clear_ready(),
            Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
            Err(error) => return Poll::Ready(Err(error)),
        }
    }
}

impl Drop for Monitor {
    fn drop(&mut self) {
        if let Err(error) = self.stop.write_all(&[1])
            && error.kind() != io::ErrorKind::BrokenPipe
        {
            eprintln!("haematite follow_monitor_stop_failed: {error}");
        }
        if let Some(worker) = self.worker.take()
            && worker.join().is_err()
        {
            eprintln!("haematite follow_monitor_panicked");
        }
    }
}

/// One head frame's result.
fn head(branch: &str, version: Version) -> Value {
    json!({"branch": branch, "version": version})
}

fn unknown(branch: &str) -> Failure {
    Failure::new(
        "branch_unknown",
        format!("branch {branch} is not published, so it has no head to follow"),
    )
}

/// The time left on a pass that expires at `expires` seconds since the epoch,
/// at `now`; `None` for a follow with no pass. A pass is good while `now` is
/// before its expiry, as lys-pass verifies it, so none left is `pass_expired`.
/// An expiry no clock can name is refused the same way.
fn remaining(expires: Option<u64>, now: SystemTime) -> Result<Option<Duration>, Failure> {
    let Some(expires) = expires else {
        return Ok(None);
    };
    UNIX_EPOCH
        .checked_add(Duration::from_secs(expires))
        .and_then(|end| end.duration_since(now).ok())
        .filter(|left| !left.is_zero())
        .map(Some)
        .ok_or_else(|| {
            Failure::new(
                "pass_expired",
                "the Lys pass this connection presented has expired; follow again with a new one",
            )
        })
}

#[cfg(test)]
#[path = "follow_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "follow_hangup_tests.rs"]
mod hangup_tests;