bothy 0.4.0

//! Background sync: debounced after local writes, periodic for pulls, backoff on errors.

use std::sync::Arc;
use std::time::Duration;

use tokio::sync::{oneshot, watch};
use tokio::time::{Instant, sleep_until};

use crate::database::{Inner, SyncReport};
use crate::error::ErrorKind;

/// Timing of background sync.
#[derive(Clone, Copy, Debug)]
pub struct AutoSync {
    /// Quiet time after a local write before syncing.
    pub debounce: Duration,
    /// Longest wait after the first unsynced local write, even if writes keep coming.
    pub max_delay: Duration,
    /// Pull interval when nothing happens locally.
    pub poll_interval: Duration,
    /// Cap of the exponential backoff after failures.
    pub max_backoff: Duration,
    /// Quiet time after a local write while a direct channel to a peer is open
    /// ([`crate::Config::p2p`]), so the peer sees it quickly.
    pub p2p_debounce: Duration,
}

/// Polling while a direct channel is being set up, to pick up the peer's answer.
const SIGNALING_POLL: Duration = Duration::from_secs(3);
/// Polls after a peer announced a segment too large to send, until the remote lists it.
const HINT_FOLLOW_UP: [Duration; 2] = [Duration::from_secs(2), Duration::from_secs(5)];

impl Default for AutoSync {
    fn default() -> Self {
        Self {
            debounce: Duration::from_secs(2),
            max_delay: Duration::from_secs(10),
            poll_interval: Duration::from_secs(30),
            max_backoff: Duration::from_secs(15 * 60),
            p2p_debounce: Duration::from_millis(100),
        }
    }
}

impl AutoSync {
    /// Settings for background or embedded use: 5-minute polling.
    pub fn background() -> Self {
        Self {
            poll_interval: Duration::from_secs(5 * 60),
            ..Self::default()
        }
    }
}

/// Outcome of the latest background sync.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum SyncStatus {
    /// No sync has finished yet.
    Idle,
    /// The last sync succeeded.
    Ok(SyncReport),
    /// The last sync failed; it will be retried after a backoff.
    Failed {
        /// Error category.
        kind: ErrorKind,
        /// Error message.
        message: String,
    },
}

/// Controls a background sync task. Dropping it stops the task.
#[derive(Debug)]
pub struct SyncHandle {
    stop: Option<oneshot::Sender<()>>,
    status: watch::Receiver<SyncStatus>,
    task: Option<tokio::task::JoinHandle<()>>,
}

impl SyncHandle {
    /// Latest status. Use [`watch::Receiver::changed`] on [`SyncHandle::watch`] to wait for updates.
    pub fn status(&self) -> SyncStatus {
        self.status.borrow().clone()
    }

    /// A receiver that is notified after every background sync.
    pub fn watch(&self) -> watch::Receiver<SyncStatus> {
        self.status.clone()
    }

    /// Stops the task and waits for a running sync to finish.
    pub async fn stop(mut self) {
        if let Some(tx) = self.stop.take() {
            let _ = tx.send(());
        }
        if let Some(task) = self.task.take() {
            let _ = task.await;
        }
    }
}

impl Drop for SyncHandle {
    fn drop(&mut self) {
        if let Some(tx) = self.stop.take() {
            let _ = tx.send(());
        }
    }
}

pub(crate) fn start(inner: Arc<Inner>, opts: AutoSync) -> SyncHandle {
    let (stop_tx, mut stop_rx) = oneshot::channel();
    let (status_tx, status_rx) = watch::channel(SyncStatus::Idle);
    let task = tokio::spawn(async move {
        #[cfg(feature = "p2p")]
        let p2p = inner
            .p2p_config
            .clone()
            .map(|config| crate::p2p::start(&inner, config));
        #[cfg(feature = "p2p")]
        let stop_p2p = |p2p: Option<tokio::task::JoinHandle<()>>| {
            let inner = inner.clone();
            async move {
                if let Some(task) = p2p {
                    crate::p2p::stop(&inner, task).await;
                }
            }
        };
        #[cfg(not(feature = "p2p"))]
        let (p2p, stop_p2p) = ((), |_: ()| async {});
        let mut backoff = Duration::ZERO;
        let mut next_poll = Instant::now();
        let mut follow_ups: Vec<Duration> = Vec::new();
        loop {
            // Wait for a local write, the poll timer, a peer's hint or a retry after a failure.
            let local = tokio::select! {
                _ = &mut stop_rx => return stop_p2p(p2p).await,
                _ = inner.poke.notified() => true,
                _ = inner.remote_hint.notified() => {
                    follow_ups = HINT_FOLLOW_UP.to_vec();
                    false
                }
                _ = sleep_until(next_poll) => false,
            };
            if local {
                let deadline = Instant::now() + opts.max_delay;
                loop {
                    let (connected, _) = inner.p2p_activity();
                    let debounce = if connected {
                        opts.p2p_debounce
                    } else {
                        opts.debounce
                    };
                    let quiet = Instant::now() + debounce;
                    tokio::select! {
                        _ = &mut stop_rx => return stop_p2p(p2p).await,
                        _ = inner.poke.notified() => {
                            if Instant::now() >= deadline { break; }
                        }
                        _ = sleep_until(quiet.min(deadline)) => break,
                    }
                }
            }
            let status = match crate::engine::sync(&inner).await {
                Ok(report) => {
                    backoff = Duration::ZERO;
                    SyncStatus::Ok(report)
                }
                Err(e) => {
                    backoff = (backoff * 2).clamp(Duration::from_secs(1), opts.max_backoff);
                    SyncStatus::Failed {
                        kind: e.kind(),
                        message: e.message().to_owned(),
                    }
                }
            };
            let _ = status_tx.send(status);
            // After a failure the next attempt waits for the backoff, not the poll interval.
            let (_, signaling) = inner.p2p_activity();
            let poll = if !follow_ups.is_empty() {
                follow_ups.remove(0)
            } else if signaling {
                opts.poll_interval.min(SIGNALING_POLL)
            } else {
                opts.poll_interval
            };
            next_poll = Instant::now() + if backoff.is_zero() { poll } else { backoff };
        }
    });
    SyncHandle {
        stop: Some(stop_tx),
        status: status_rx,
        task: Some(task),
    }
}