ihe-atna 0.4.0

// SPDX-FileCopyrightText: Cadasto B.V.
// SPDX-License-Identifier: BUSL-1.1

//! The store-and-forward sender of ITI-20 messages (ITI TF-2 §3.20.4.1.1).
//!
//! Every message is stored in the [`Spool`] first and delivered from it,
//! oldest first, so a message counts as recorded once it is stored, and a
//! repository that cannot be reached delays its delivery without losing it.
//! [`Forwarder::submit`] only ever writes to the spool: it never waits on
//! the network, so a slow repository never holds an exchange, and it waits
//! on the disk at most the spool's [`Bounds::write_timeout`](super::spool::Bounds::write_timeout).
//! A message stored after its exchange stopped waiting for it is delivered
//! like any other.
//!
//! [`Forwarder::run`] is the delivery loop. To a syslog repository it keeps
//! one connection open, writes each stored message to it within the
//! repository's timeouts, removes the message once it is written, and after
//! any transport failure drops the connection and retries with an
//! exponential backoff, with jitter, capped at the configured longest wait.
//! A connection the repository ended, and the open connection when
//! [`Forwarder::stop`] ends the loop, are closed with
//! [`Connection::close`], so the frames written last are not lost to a
//! reset (RFC 5425 §4.4).
//! To a FHIR Feed repository (feature `balp`) it posts each stored record,
//! removes it on a `2xx`, retries the same way after a failure that may
//! pass, and quarantines a record the repository refused, so the drain goes
//! on.

use std::fmt;
use std::hash::BuildHasher as _;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::{Duration, Instant};

use secrecy::SecretSlice;
use tokio::sync::Notify;

use super::chain;
#[cfg(feature = "balp")]
use super::feed::{FeedError, FeedRepository};
#[cfg(feature = "atna")]
use super::repository::{Connection, Repository};
use super::spool::reserve::Reservation;
use super::spool::{Depth, Spool, SpoolError, Stored};

/// The first wait after a failed delivery.
const FIRST_RETRY: Duration = Duration::from_millis(250);

/// What the forwarder reports of itself.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Status {
    /// What the store holds, awaiting delivery or in quarantine.
    pub depth: Depth,
    /// The messages delivered since the forwarder started.
    pub delivered: u64,
    /// The failed delivery attempts since the forwarder started, each
    /// followed by a backoff.
    pub retries: u64,
    /// Whether the last attempt to deliver succeeded; `false` while the
    /// forwarder retries.
    pub reachable: bool,
    /// Whether the store keeps its messages across a restart.
    pub durable: bool,
    /// The messages the store refused for want of room since it opened.
    pub refused: u64,
}

/// A durable store a forwarder delivers from, oldest first: the crate's
/// [`Spool`], or a caller's own, such as a table of records not yet
/// delivered in the caller's database.
///
/// ITI TF-2 §3.20.4.1.1 has a sender store a record it cannot deliver and
/// send it when it can; where it stores it is the sender's. The forwarder
/// asks for the oldest message, delivers it, and then removes it, or moves it
/// aside when the repository refused it, so the store decides nothing of the
/// delivery.
#[async_trait::async_trait]
pub trait Outbox: Send + Sync + fmt::Debug {
    /// The oldest message awaiting delivery, or `None` when there is none.
    ///
    /// # Errors
    ///
    /// An [`OutboxError`] when the store cannot be read; the forwarder backs
    /// off and asks again.
    async fn oldest(&self) -> Result<Option<Stored>, OutboxError>;

    /// Returns once a message may have been stored since the last
    /// [`Outbox::oldest`] found none, or after a while, so the forwarder asks
    /// again.
    async fn stored(&self);

    /// Removes the delivered message `sequence`.
    ///
    /// # Errors
    ///
    /// An [`OutboxError`] when it cannot be removed; it is then delivered
    /// again, which a syslog or FHIR Feed repository takes as a second copy.
    async fn remove(&self, sequence: u64) -> Result<(), OutboxError>;

    /// Moves the message `sequence`, which the repository refused, out of
    /// the way of the drain, keeping it for an operator.
    ///
    /// # Errors
    ///
    /// An [`OutboxError`] when it cannot be moved.
    async fn quarantine(&self, sequence: u64) -> Result<(), OutboxError>;

    /// What the store holds now.
    fn depth(&self) -> Depth;

    /// Whether the store keeps its messages across a restart.
    fn is_durable(&self) -> bool;

    /// The messages the store refused for want of room since it opened.
    fn refusals(&self) -> u64 {
        0
    }
}

/// Why an [`Outbox`] could not do what the forwarder asked.
#[derive(Debug, thiserror::Error)]
#[error("the audit store could not be read or written")]
pub struct OutboxError(#[source] pub Box<dyn std::error::Error + Send + Sync>);

#[async_trait::async_trait]
impl<O: Outbox + ?Sized> Outbox for Arc<O> {
    async fn oldest(&self) -> Result<Option<Stored>, OutboxError> {
        O::oldest(self).await
    }

    async fn stored(&self) {
        O::stored(self).await;
    }

    async fn remove(&self, sequence: u64) -> Result<(), OutboxError> {
        O::remove(self, sequence).await
    }

    async fn quarantine(&self, sequence: u64) -> Result<(), OutboxError> {
        O::quarantine(self, sequence).await
    }

    fn depth(&self) -> Depth {
        O::depth(self)
    }

    fn is_durable(&self) -> bool {
        O::is_durable(self)
    }

    fn refusals(&self) -> u64 {
        O::refusals(self)
    }
}

#[async_trait::async_trait]
impl Outbox for Spool {
    async fn oldest(&self) -> Result<Option<Stored>, OutboxError> {
        Spool::oldest(self)
            .await
            .map_err(|error| OutboxError(Box::new(error)))
    }

    async fn stored(&self) {
        Spool::stored(self).await;
    }

    async fn remove(&self, sequence: u64) -> Result<(), OutboxError> {
        Spool::remove(self, sequence)
            .await
            .map_err(|error| OutboxError(Box::new(error)))
    }

    async fn quarantine(&self, sequence: u64) -> Result<(), OutboxError> {
        self.reject(sequence)
            .await
            .map_err(|error| OutboxError(Box::new(error)))
    }

    fn depth(&self) -> Depth {
        Spool::depth(self)
    }

    fn is_durable(&self) -> bool {
        Spool::is_durable(self)
    }

    fn refusals(&self) -> u64 {
        self.refused()
    }
}

/// Where a forwarder delivers to.
#[derive(Debug)]
enum Destination {
    /// A syslog repository (ITI TF-2 §3.20.4.1).
    #[cfg(feature = "atna")]
    Syslog(Repository),
    /// A FHIR Feed repository (ITI TF-2 §3.20.4.2).
    #[cfg(feature = "balp")]
    Feed(FeedRepository),
}

/// The store-and-forward sender to one Audit Record Repository, from the
/// crate's [`Spool`] or from a caller's [`Outbox`].
#[derive(Debug)]
pub struct Forwarder<O: Outbox = Spool> {
    outbox: O,
    destination: Destination,
    retry_max: Duration,
    delivered: AtomicU64,
    retries: AtomicU64,
    reachable: AtomicBool,
    stopping: AtomicBool,
    stop: Notify,
}

impl Forwarder<Spool> {
    /// The sender of the syslog frames in `spool` to `repository`, waiting
    /// at most `retry_max` between two attempts.
    #[cfg(feature = "atna")]
    #[must_use]
    pub fn new(spool: Spool, repository: Repository, retry_max: Duration) -> Arc<Self> {
        Self::to(spool, Destination::Syslog(repository), retry_max)
    }

    /// The sender of the `AuditEvent` records in `spool` to the FHIR Feed
    /// repository `feed`, waiting at most `retry_max` between two attempts.
    #[cfg(feature = "balp")]
    #[must_use]
    pub fn fhir_feed(spool: Spool, feed: FeedRepository, retry_max: Duration) -> Arc<Self> {
        Self::to(spool, Destination::Feed(feed), retry_max)
    }

    /// Stores `frame` for delivery, returning once it is stored.
    ///
    /// # Errors
    ///
    /// The [`SpoolError`] of a spool that is full or cannot be written: the
    /// message is then neither stored nor delivered. Past the spool's
    /// [`Bounds::write_timeout`](super::spool::Bounds::write_timeout),
    /// [`SpoolError::Late`]: the message is not stored yet, and is delivered
    /// once its write stores it.
    pub async fn submit(&self, frame: SecretSlice<u8>) -> Result<(), SpoolError> {
        self.outbox.push(frame).await
    }

    /// Takes the place of one message of at most `bytes` bytes in the spool
    /// before the message exists, as [`Spool::reserve`] does.
    ///
    /// # Errors
    ///
    /// [`SpoolError::Full`] when the spool has no room for it.
    pub fn reserve(&self, bytes: u64) -> Result<Reservation, SpoolError> {
        self.outbox.reserve(bytes)
    }
}

impl<O: Outbox> Forwarder<O> {
    /// The sender of the syslog frames in the caller's `outbox` to
    /// `repository`, waiting at most `retry_max` between two attempts.
    #[cfg(feature = "atna")]
    #[must_use]
    pub fn syslog_from(outbox: O, repository: Repository, retry_max: Duration) -> Arc<Self> {
        Self::to(outbox, Destination::Syslog(repository), retry_max)
    }

    /// The sender of the `AuditEvent` records in the caller's `outbox` to the
    /// FHIR Feed repository `feed`, waiting at most `retry_max` between two
    /// attempts.
    #[cfg(feature = "balp")]
    #[must_use]
    pub fn fhir_feed_from(outbox: O, feed: FeedRepository, retry_max: Duration) -> Arc<Self> {
        Self::to(outbox, Destination::Feed(feed), retry_max)
    }

    fn to(outbox: O, destination: Destination, retry_max: Duration) -> Arc<Self> {
        Arc::new(Self {
            outbox,
            destination,
            retry_max: retry_max.max(FIRST_RETRY),
            delivered: AtomicU64::new(0),
            retries: AtomicU64::new(0),
            reachable: AtomicBool::new(true),
            stopping: AtomicBool::new(false),
            stop: Notify::new(),
        })
    }

    /// Asks [`Forwarder::run`] to end: it finishes the message it is
    /// writing, if any, closes its syslog connection with
    /// [`Connection::close`], and returns, so the caller awaits the task it
    /// spawned.
    ///
    /// A message not yet written stays stored and is delivered by the next
    /// run. A run started after the stop returns at once.
    pub fn stop(&self) {
        self.stopping.store(true, Ordering::SeqCst);
        self.stop.notify_waiters();
    }

    /// Returns once [`Forwarder::stop`] has been called.
    async fn stopped(&self) {
        loop {
            // NOTE: a Notified future receives notify_waiters from the moment it
            // is created (tokio::sync::Notify), so no stop falls between the two.
            let notified = self.stop.notified();
            if self.stopping.load(Ordering::SeqCst) {
                return;
            }
            notified.await;
        }
    }

    /// What the forwarder holds and how its deliveries went.
    #[must_use]
    pub fn status(&self) -> Status {
        Status {
            depth: self.outbox.depth(),
            delivered: self.delivered.load(Ordering::Relaxed),
            retries: self.retries.load(Ordering::Relaxed),
            reachable: self.reachable.load(Ordering::Relaxed),
            durable: self.outbox.is_durable(),
            refused: self.outbox.refusals(),
        }
    }

    /// Delivers the stored messages, oldest first, until
    /// [`Forwarder::stop`] is called.
    ///
    /// The caller spawns it once and stops it with [`Forwarder::stop`],
    /// awaiting the task, which closes the syslog connection first. A task
    /// aborted instead drops the connection, and the frames written last
    /// may then be lost to a reset (RFC 5425 §4.4); a message in flight
    /// stays stored and is delivered again by the next run.
    pub async fn run(self: Arc<Self>) {
        match &self.destination {
            #[cfg(feature = "atna")]
            Destination::Syslog(repository) => self.run_syslog(repository).await,
            #[cfg(feature = "balp")]
            Destination::Feed(feed) => self.run_feed(feed).await,
        }
    }

    /// The oldest stored message, after waiting for one while the spool is
    /// empty; `None` after a wake-up or after a failure to read the spool,
    /// which has backed off.
    async fn next(&self, retry: &mut Duration) -> Option<Stored> {
        match self.outbox.oldest().await {
            Ok(Some(stored)) => Some(stored),
            Ok(None) => {
                self.outbox.stored().await;
                None
            }
            Err(error) => {
                tracing::warn!(target: crate::atna::targets::FORWARDER, error = %chain(&error), "the audit spool could not be read");
                *retry = self.back_off(*retry).await;
                None
            }
        }
    }

    /// Removes the delivered message `sequence` and counts it.
    async fn delivered(&self, sequence: u64, retry: &mut Duration) {
        self.reachable.store(true, Ordering::Relaxed);
        *retry = FIRST_RETRY;
        match self.outbox.remove(sequence).await {
            Ok(()) => {
                self.delivered.fetch_add(1, Ordering::Relaxed);
            }
            Err(error) => {
                tracing::warn!(target: crate::atna::targets::FORWARDER,
                    error = %chain(&error),
                    "a delivered audit message could not be removed from the spool"
                );
                *retry = self.back_off(*retry).await;
            }
        }
    }

    // NOTE: RFC 5425 §4.3 carries no application-level acknowledgement, so a
    // syslog repository never refuses a frame: only a transport failure retries.
    #[cfg(feature = "atna")]
    async fn run_syslog(&self, repository: &Repository) {
        let mut connection: Option<Connection> = None;
        let mut retry = FIRST_RETRY;
        loop {
            let next = tokio::select! {
                biased;
                () = self.stopped() => break,
                next = self.next(&mut retry) => next,
            };
            let Some(stored) = next else {
                continue;
            };
            let reusable = match connection.take() {
                Some(mut open) => {
                    if open.is_closed().await {
                        closed(open).await;
                        None
                    } else {
                        Some(open)
                    }
                }
                None => None,
            };
            let mut open = match reusable {
                Some(open) => open,
                None => match repository.connect().await {
                    Ok(open) => open,
                    Err(error) => {
                        tracing::warn!(target: crate::atna::targets::FORWARDER,
                            error = %chain(&error),
                            "the audit repository could not be reached; its messages stay spooled"
                        );
                        retry = self.back_off(retry).await;
                        continue;
                    }
                },
            };
            if let Err(error) = open.send(&stored.message).await {
                tracing::warn!(target: crate::atna::targets::FORWARDER,
                    error = %chain(&error),
                    "an audit message could not be delivered; the connection is dropped and the message stays spooled"
                );
                retry = self.back_off(retry).await;
                continue;
            }
            connection = Some(open);
            self.delivered(stored.sequence, &mut retry).await;
        }
        if let Some(open) = connection {
            closed(open).await;
        }
    }

    // NOTE: the `RESTful` ATNA supplement §3.20.4.3.3 leaves a failure to the
    // client: a refusal is quarantined, every other failure is retried.
    #[cfg(feature = "balp")]
    async fn run_feed(&self, feed: &FeedRepository) {
        let mut retry = FIRST_RETRY;
        loop {
            let next = tokio::select! {
                biased;
                () = self.stopped() => break,
                next = self.next(&mut retry) => next,
            };
            let Some(stored) = next else {
                continue;
            };
            match feed.send(&stored.message).await {
                Ok(()) => self.delivered(stored.sequence, &mut retry).await,
                Err(FeedError::Rejected { status }) => {
                    self.reachable.store(true, Ordering::Relaxed);
                    tracing::error!(target: crate::atna::targets::FORWARDER,
                        sequence = stored.sequence,
                        status = status.as_u16(),
                        "the audit repository refused an AuditEvent; it is quarantined"
                    );
                    if let Err(error) = self.outbox.quarantine(stored.sequence).await {
                        tracing::warn!(target: crate::atna::targets::FORWARDER,
                            error = %chain(&error),
                            "a refused audit message could not be quarantined"
                        );
                        retry = self.back_off(retry).await;
                    }
                }
                Err(error) => {
                    tracing::warn!(target: crate::atna::targets::FORWARDER,
                        error = %chain(&error),
                        "an AuditEvent could not be delivered; it stays spooled"
                    );
                    retry = self.back_off(retry).await;
                }
            }
        }
    }

    /// Records a failed attempt, waits about `retry`, and returns the wait
    /// after it: twice as long, at most the configured longest wait.
    /// A stop ends the wait early.
    async fn back_off(&self, retry: Duration) -> Duration {
        self.reachable.store(false, Ordering::Relaxed);
        self.retries.fetch_add(1, Ordering::Relaxed);
        tokio::select! {
            () = self.stopped() => {}
            () = tokio::time::sleep(jittered(retry)) => {}
        }
        retry.saturating_mul(2).min(self.retry_max)
    }
}

/// Closes `connection`, on which no further frame follows, with
/// [`Connection::close`], and logs a close that fails: a frame written last
/// may then be lost.
#[cfg(feature = "atna")]
async fn closed(connection: Connection) {
    if let Err(error) = connection.close().await {
        tracing::warn!(target: crate::atna::targets::FORWARDER,
            error = %chain(&error),
            "the audit repository connection did not close cleanly; the messages written last may be lost"
        );
    }
}

/// A wait between half of `retry` and `retry`, so senders that failed
/// together do not retry together (no specification governs this: our own
/// design).
fn jittered(retry: Duration) -> Duration {
    let half = retry / 2;
    let spread = u64::try_from(half.as_nanos()).unwrap_or(u64::MAX);
    let random = std::collections::hash_map::RandomState::new().hash_one(Instant::now());
    half.saturating_add(Duration::from_nanos(random % spread.saturating_add(1)))
}

#[cfg(all(test, feature = "atna"))]
mod tests {
    use std::sync::Arc;
    use std::time::{Duration, Instant};

    use secrecy::{ExposeSecret as _, SecretSlice};
    use tokio::io::AsyncReadExt as _;
    use tokio::net::TcpListener;
    use url::Url;

    use super::super::repository::{Repository, Timeouts};
    use super::super::spool::in_flight::tests::{
        BOUND, Logs, PATIENT, SLACK, Stall, naming_the_patient,
    };
    use super::super::spool::{Bounds, Spool, SpoolError};
    use super::Forwarder;

    /// ITI TF-2 §3.20.4.1.1: a message its exchange gave up on, stored once
    /// the stalled write ends, reaches the repository like any other.
    #[tokio::test]
    async fn a_message_stored_after_its_exchange_gave_up_is_delivered_once() {
        let logs = Logs::default();
        let _logging = tracing::subscriber::set_default(logs.subscriber());
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("a listener");
        let url = Url::parse(&format!(
            "tcp://{}",
            listener.local_addr().expect("an address")
        ))
        .expect("a URL");
        let timeouts = Timeouts {
            connect: SLACK,
            send: SLACK,
        };
        let repository =
            Repository::unencrypted_for_development(&url, timeouts).expect("a repository");
        let directory = tempfile::tempdir().expect("a directory");
        let bounds = Bounds {
            max_messages: 16,
            max_bytes: 1 << 20,
            write_timeout: BOUND,
        };
        let spool = Spool::open(&directory.path().join("spool"), bounds).expect("it opens");
        let forwarder = Forwarder::new(spool.clone(), repository, Duration::from_millis(400));
        let running = tokio::spawn(Arc::clone(&forwarder).run());
        // The forwarder finds the spool empty and waits for a stored message.
        tokio::time::sleep(Duration::from_millis(50)).await;
        let stall = Stall::of(&spool);

        let asked = Instant::now();
        let refused = forwarder.submit(naming_the_patient()).await;
        assert!(asked.elapsed() < BOUND + SLACK, "{:?}", asked.elapsed());
        assert!(
            matches!(refused, Err(SpoolError::Late { .. })),
            "{refused:?}"
        );
        assert_eq!(
            0,
            forwarder.status().depth.messages,
            "not counted as stored"
        );

        stall.release();
        let (mut stream, _) = tokio::time::timeout(SLACK, listener.accept())
            .await
            .expect("the forwarder connects once the message is stored")
            .expect("a connection");
        let sent = naming_the_patient();
        let mut received = vec![0_u8; sent.expose_secret().len()];
        tokio::time::timeout(SLACK, stream.read_exact(&mut received))
            .await
            .expect("the message arrives")
            .expect("it reads");
        assert_eq!(sent.expose_secret(), received.as_slice());

        let until = Instant::now() + SLACK;
        while forwarder.status().delivered == 0 && Instant::now() < until {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        let status = forwarder.status();
        assert_eq!(1, status.delivered, "delivered once");
        assert_eq!(0, status.depth.messages);
        running.abort();
        assert!(!logs.text().contains(PATIENT), "{}", logs.text());
    }

    /// The RFC 5425 frame of `text`.
    fn framed(text: &str) -> SecretSlice<u8> {
        SecretSlice::from(format!("{} {text}", text.len()).into_bytes())
    }

    /// ITI TF-2 §3.20.4.1.1: the record of a failed exchange whose caller
    /// stopped waiting while it was still queued behind another write is
    /// written once that write ends, and delivered once.
    #[tokio::test]
    async fn a_record_given_up_while_queued_is_stored_and_delivered_once() {
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("a listener");
        let url = Url::parse(&format!(
            "tcp://{}",
            listener.local_addr().expect("an address")
        ))
        .expect("a URL");
        let timeouts = Timeouts {
            connect: SLACK,
            send: SLACK,
        };
        let repository =
            Repository::unencrypted_for_development(&url, timeouts).expect("a repository");
        let directory = tempfile::tempdir().expect("a directory");
        let bounds = Bounds {
            max_messages: 16,
            max_bytes: 1 << 20,
            write_timeout: BOUND,
        };
        let spool = Spool::open(&directory.path().join("spool"), bounds).expect("it opens");
        let forwarder = Forwarder::new(spool.clone(), repository, Duration::from_millis(400));
        let running = tokio::spawn(Arc::clone(&forwarder).run());
        // The forwarder finds the spool empty and waits for a stored message.
        tokio::time::sleep(Duration::from_millis(50)).await;
        let stall = Stall::of(&spool);

        let holding = forwarder
            .submit(framed("the write that holds the lock"))
            .await;
        assert!(
            matches!(holding, Err(SpoolError::Late { .. })),
            "{holding:?}"
        );
        let gave_up = tokio::time::timeout(
            BOUND / 4,
            forwarder.submit(framed("the failed exchange's record")),
        )
        .await;
        assert!(
            gave_up.is_err(),
            "the exchange's time ran out while its record was queued"
        );
        assert_eq!(0, forwarder.status().depth.messages, "nothing stored yet");

        stall.release();
        let (mut stream, _) = tokio::time::timeout(SLACK, listener.accept())
            .await
            .expect("the forwarder connects once a message is stored")
            .expect("a connection");
        for text in [
            "the write that holds the lock",
            "the failed exchange's record",
        ] {
            let sent = framed(text);
            let mut received = vec![0_u8; sent.expose_secret().len()];
            tokio::time::timeout(SLACK, stream.read_exact(&mut received))
                .await
                .expect("the message arrives")
                .expect("it reads");
            assert_eq!(
                sent.expose_secret(),
                received.as_slice(),
                "in the order they were queued"
            );
        }
        let until = Instant::now() + SLACK;
        while forwarder.status().delivered < 2 && Instant::now() < until {
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        let mut more = [0_u8; 1];
        let extra = tokio::time::timeout(BOUND, stream.read(&mut more)).await;
        assert!(extra.is_err(), "nothing is delivered twice: {extra:?}");
        let status = forwarder.status();
        assert_eq!(2, status.delivered, "each delivered once");
        assert_eq!(0, status.depth.messages);
        running.abort();
    }
}