taktwerk 0.6.0

//! The puller's bookkeeping, free of I/O: where the next read starts, which cycles or events
//! were lost, how fast to poll.

use std::time::Duration;

use taktwerk_core::record::wire::{MAX_PULL_BYTES, row_bytes};

/// Rows asked for per `read_rows` call at most.
pub const MAX_ROWS: u32 = 1 << 20;

/// Events asked for per `read_events` call.
pub const MAX_EVENTS: u32 = 4096;

/// Rows lost between two pulled rows: cycles `first..=last`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Lost {
    /// First missing cycle.
    pub first: u64,
    /// Last missing cycle.
    pub last: u64,
    /// Rows missing.
    pub rows: u64,
}

impl Lost {
    /// The `detail` of its gap event.
    #[must_use]
    pub fn detail(&self) -> String {
        format!(
            "cycles {}..={} lost ({} rows)",
            self.first, self.last, self.rows
        )
    }
}

/// What to do with a pulled row.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Verdict {
    /// Write it.
    Write,
    /// Write it; rows before it were lost.
    AfterGap(Lost),
    /// Already written; skip it.
    Skip,
}

/// Position of the puller in one engine run's rows.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Rows {
    every: u64,
    /// `from_cycle` of the next read; 0 (from the oldest row held) until the first row.
    from: u64,
    /// Cycle of the next row expected; `None` before the first.
    expected: Option<u64>,
}

impl Rows {
    /// Start of a run, one row every `every` ticks.
    #[must_use]
    pub fn new(every: u32) -> Self {
        Self {
            every: u64::from(every.max(1)),
            from: 0,
            expected: None,
        }
    }

    /// `from_cycle` of the next read.
    #[must_use]
    pub const fn from(&self) -> u64 {
        self.from
    }

    /// Account for the row of `cycle`.
    pub fn row(&mut self, cycle: u64) -> Verdict {
        let verdict = match self.expected {
            Some(e) if cycle < e => return Verdict::Skip,
            Some(e) if cycle > e => {
                let last = cycle - self.every;
                Verdict::AfterGap(Lost {
                    first: e,
                    last,
                    rows: (last - e) / self.every + 1,
                })
            }
            _ => Verdict::Write,
        };
        self.expected = Some(cycle + self.every);
        self.from = self.from.max(cycle + self.every);
        verdict
    }

    /// A read returned `next_cycle`.
    pub fn advance(&mut self, next_cycle: u64) {
        self.from = self.from.max(next_cycle);
    }
}

/// Position of the puller in one engine run's event log.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct Events {
    from: u64,
    started: bool,
}

impl Events {
    /// `from_seq` of the next read.
    #[must_use]
    pub const fn from(&self) -> u64 {
        self.from
    }

    /// A read returned `oldest_seq` and `next_seq`; the events lost since the last read,
    /// `first..=last`. Events gone before the first read are not counted.
    pub fn read(&mut self, oldest_seq: u64, next_seq: u64) -> Option<(u64, u64)> {
        let lost = (self.started && self.from < oldest_seq).then(|| (self.from, oldest_seq - 1));
        self.started = true;
        self.from = self.from.max(next_seq);
        lost
    }
}

/// Rows one `read_rows` call can return for rows of `width` words.
#[must_use]
pub fn rows_per_call(width: usize) -> u32 {
    let fit = (MAX_PULL_BYTES / row_bytes(width)).max(1);
    u32::try_from(fit).unwrap_or(MAX_ROWS).min(MAX_ROWS)
}

/// Pause between reads that came back short: a tenth of the engine's buffer, at most 1 s and at
/// least 10 ms.
#[must_use]
pub fn poll_period(buffer_s: f64) -> Duration {
    let s = if buffer_s.is_finite() {
        (buffer_s / 10.0).clamp(0.01, 1.0)
    } else {
        1.0
    };
    Duration::from_secs_f64(s)
}

/// Whether to read again at once: a batch came back full.
#[must_use]
pub const fn more(returned: usize, asked: u32) -> bool {
    returned as u64 >= asked as u64
}

/// What one `taktwerk record` session did.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct Summary {
    /// Runs opened.
    pub runs: u64,
    /// Rows written.
    pub rows: u64,
    /// Parts written.
    pub parts: u64,
    /// Gaps logged, rows and events.
    pub gaps: u64,
}

impl std::fmt::Display for Summary {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "{} rows, {} gaps, {} parts in {} runs",
            self.rows, self.gaps, self.parts, self.runs
        )
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn rows_continue_from_the_last_and_report_jumps() {
        let mut r = Rows::new(2);
        assert_eq!(r.from(), 0);
        r.advance(0);
        assert_eq!(
            r.from(),
            0,
            "an empty first read keeps asking from the oldest"
        );
        assert_eq!(r.row(10), Verdict::Write, "the first row is never a gap");
        assert_eq!(r.row(12), Verdict::Write);
        r.advance(14);
        assert_eq!(r.from(), 14);
        assert_eq!(
            r.row(20),
            Verdict::AfterGap(Lost {
                first: 14,
                last: 18,
                rows: 3
            })
        );
        assert_eq!(r.row(20), Verdict::Skip);
        assert_eq!(r.row(22), Verdict::Write);
        r.advance(10);
        assert_eq!(r.from(), 24, "never backwards");
        assert_eq!(
            Lost {
                first: 14,
                last: 18,
                rows: 3
            }
            .detail(),
            "cycles 14..=18 lost (3 rows)"
        );
    }

    #[test]
    fn event_loss_counts_only_after_the_first_read() {
        let mut e = Events::default();
        assert_eq!(
            e.read(5, 9),
            None,
            "events gone before we came are not lost"
        );
        assert_eq!(e.from(), 9);
        assert_eq!(e.read(5, 9), None);
        assert_eq!(e.read(12, 20), Some((9, 11)));
        assert_eq!(e.from(), 20);
    }

    #[test]
    fn batches_and_polling_follow_the_buffer() {
        assert_eq!(rows_per_call(2), u32::try_from((8 << 20) / 24).unwrap_or(0));
        assert_eq!(rows_per_call(0), MAX_ROWS);
        assert_eq!(rows_per_call(10 << 20), 1);
        assert_eq!(poll_period(600.0), Duration::from_secs(1));
        assert_eq!(poll_period(2.0), Duration::from_millis(200));
        assert_eq!(poll_period(0.0), Duration::from_millis(10));
        assert_eq!(poll_period(f64::NAN), Duration::from_secs(1));
        assert!(more(10, 10));
        assert!(!more(9, 10));
    }
}