haematite-cli 0.21.3

//! Resolve instant bounds once against the captured branch's retained clock.

use std::num::NonZeroUsize;

use haematite_query::stage::window::Window;
use serde_json::value::RawValue;

use super::super::parser::window::{Bound, Window as AskedWindow};
use super::Context;
use crate::serve::protocol::Version;
use crate::serve::storage::{Captured, Failure};

pub(in crate::serve) fn resolve(
    context: &Context<'_>,
    at: &Captured,
    now: &RawValue,
    scan: u64,
    asked: &AskedWindow,
) -> Result<Window, Failure> {
    let successor = || {
        at.version.sequence.checked_add(1).ok_or_else(|| {
            Failure::new(
                "query_window_bound",
                "captured version has no exclusive successor",
            )
        })
    };
    let (from, to) = match (asked.from, asked.to) {
        (Bound::Version(from), Bound::Version(to)) => (from, to),
        (Bound::Version(from), Bound::Now) => (from, successor()?),
        (Bound::Instant(from), Bound::Instant(_) | Bound::Now) => {
            let to = match asked.to {
                Bound::Instant(to) => to,
                Bound::Now => clock(now)?,
                Bound::Version(_) => {
                    return Err(Failure::new(
                        "query_window_bound",
                        "window mixes versions and instants",
                    ));
                }
            };
            if from > to {
                return Err(reversed());
            }
            let maximum = usize::try_from(scan)
                .ok()
                .and_then(NonZeroUsize::new)
                .ok_or_else(|| {
                    Failure::new(
                        "query_window_history_limit",
                        "window needs a representable nonzero history scan allowance",
                    )
                })?;
            let history = context.catalogue.window_clock(at, maximum)?;
            (
                version_at(&history, at.version, from)?,
                version_at(&history, at.version, to)?,
            )
        }
        _ => {
            return Err(Failure::new(
                "query_window_bound",
                "window uses two versions or two RFC 3339 instants",
            ));
        }
    };
    if from > to {
        return Err(reversed());
    }
    Window::new(at.version.generation, from, to)
        .map_err(|error| Failure::new("query_window_bound", error.to_string()))
}

fn reversed() -> Failure {
    Failure::new("query_window_reversed", "window from is after to")
}

fn clock(now: &RawValue) -> Result<u64, Failure> {
    let invalid = || {
        Failure::new(
            "query_window_instant",
            "$now cannot be represented as stored epoch nanoseconds",
        )
    };
    let (seconds, nanos) = now.get().split_once('.').ok_or_else(invalid)?;
    if nanos.len() != 9
        || !seconds.bytes().all(|byte| byte.is_ascii_digit())
        || !nanos.bytes().all(|byte| byte.is_ascii_digit())
    {
        return Err(invalid());
    }
    seconds
        .parse::<u64>()
        .ok()
        .and_then(|seconds| seconds.checked_mul(1_000_000_000))
        .and_then(|seconds| {
            nanos
                .parse::<u64>()
                .ok()
                .and_then(|nanos| seconds.checked_add(nanos))
        })
        .ok_or_else(invalid)
}

/// A boundary is the first retained version at or after its instant.
pub(in crate::serve) fn version_at(
    history: &[(Version, u64)],
    at: Version,
    instant: u64,
) -> Result<u64, Failure> {
    let (first, first_time) = history.first().ok_or_else(|| {
        Failure::new(
            "query_window_history_missing",
            "window branch has no retained history clock",
        )
    })?;
    if first.sequence != 0 && instant <= *first_time {
        return Err(Failure::new(
            "query_window_history_released",
            "window instant may refer to released history",
        ));
    }
    let mut previous = None;
    let mut boundary = None;
    for (version, time) in history {
        if version.generation != at.generation
            || version.sequence > at.sequence
            || previous.is_some_and(|(sequence, timestamp)| {
                sequence >= version.sequence || timestamp > *time
            })
        {
            return Err(Failure::new(
                "query_window_history_clock",
                "window history versions or instants are not ordered",
            ));
        }
        previous = Some((version.sequence, *time));
        if boundary.is_none() && *time >= instant {
            boundary = Some(version.sequence);
        }
    }
    boundary.map_or_else(
        || {
            at.sequence.checked_add(1).ok_or_else(|| {
                Failure::new(
                    "query_window_bound",
                    "captured version has no exclusive successor",
                )
            })
        },
        Ok,
    )
}

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