use std::future::Future;
use crate::error::{Error, Result};
use crate::request::CandleRequest;
use crate::types::{Candle, Timestamp};
pub(crate) const MAX_CALLS: u32 = 100;
pub(crate) async fn read<F, Fut>(
request: &CandleRequest,
exchange: &'static str,
max_per_call: u32,
fetch: F,
) -> Result<Vec<Candle>>
where
F: Fn(Option<Timestamp>, u32) -> Fut,
Fut: Future<Output = Result<Vec<Candle>>>,
{
let interval = request.interval;
read_on_grid(
request,
exchange,
max_per_call,
move |at, count| interval.advance(at, count),
fetch,
)
.await
}
pub(crate) async fn read_on_grid<F, Fut, G>(
request: &CandleRequest,
exchange: &'static str,
max_per_call: u32,
advance: G,
fetch: F,
) -> Result<Vec<Candle>>
where
F: Fn(Option<Timestamp>, u32) -> Fut,
Fut: Future<Output = Result<Vec<Candle>>>,
G: Fn(Timestamp, i64) -> Option<Timestamp> + Copy,
{
if request.limit == Some(0) {
return Err(Error::invalid_request(
"limit",
"asking for no candles is not a request; leave `limit` unset for a default page",
));
}
if let (Some(from), Some(to)) = (request.from, request.to)
&& from >= to
{
return Err(Error::invalid_request("from", "must be earlier than `to`"));
}
let (target, field) = match (request.limit, request.from) {
(Some(limit), _) => (limit as usize, "limit"),
(None, Some(from)) => (window_candles(request, from, advance), "from"),
(None, None) => (max_per_call as usize, "limit"),
};
let ceiling = max_per_call as usize * MAX_CALLS as usize;
if target > ceiling {
return Err(Error::invalid_request(
field,
format!(
"that window is about {target} {:?} candles, and {exchange} serves {max_per_call} \
per call; `maxt` walks at most {MAX_CALLS} calls, so {ceiling} is the most one \
request can read. Ask for a narrower window and page it yourself.",
request.interval
),
));
}
if let (Some(from), Some(limit)) = (request.from, request.limit) {
if request.to.is_none() {
advance(from, i64::from(limit)).ok_or_else(|| {
Error::invalid_request(
"from",
format!(
"{limit} {:?} candles from {from} runs past the last instant a \
`Timestamp` can hold, around the year 2262",
request.interval
),
)
})?;
}
return read_from_limit(
exchange,
max_per_call,
from,
request.to.unwrap_or_else(Timestamp::now),
limit,
advance,
&fetch,
)
.await;
}
let mut cursor = request.to;
let mut collected: Vec<Candle> = Vec::new();
let mut previous_oldest: Option<Timestamp> = None;
let mut calls: u32 = 0;
loop {
let wanted = if request.from.is_some() {
target
} else {
target.saturating_sub(collected.len())
};
let count = u32::try_from(wanted)
.unwrap_or(max_per_call)
.min(max_per_call);
if count == 0 {
break;
}
if calls >= MAX_CALLS {
return Err(Error::exchange(
exchange,
"candle_pagination_ceiling",
format!(
"walked {MAX_CALLS} pages of {exchange} candles and gathered {} of the \
{target} asked for; each page carried fewer candles the walk had not \
already seen than it asked for",
collected.len()
),
));
}
calls += 1;
let mut page = fetch(cursor, count).await?;
let page_len = page.len();
let Some(oldest) = page.iter().map(|candle| candle.open_time).min() else {
break;
};
let prior = previous_oldest;
let progressed = prior.is_none_or(|previous| oldest < previous);
if !progressed {
if page_len >= count as usize {
return Err(Error::exchange(
exchange,
"candle_pagination_stalled",
"the candle cursor did not move backwards between pages",
));
}
break;
}
previous_oldest = Some(oldest);
page.retain(|candle| {
prior.is_none_or(|prior| candle.open_time < prior)
&& request.from.is_none_or(|from| candle.open_time >= from)
&& request.to.is_none_or(|to| candle.open_time < to)
});
collected.extend(page);
let short_page = page_len < count as usize;
let reached_start = request.from.is_some_and(|from| oldest <= from);
let counted_out = request.from.is_none() && collected.len() >= target;
if counted_out || short_page || reached_start {
break;
}
cursor = Some(oldest);
}
collected.sort_by_key(|candle| candle.open_time);
collected.dedup_by_key(|candle| candle.open_time);
if request.from.is_none() {
collected.drain(..collected.len().saturating_sub(target));
}
Ok(collected)
}
async fn read_from_limit<F, Fut, G>(
exchange: &'static str,
max_per_call: u32,
from: Timestamp,
upper: Timestamp,
limit: u32,
advance: G,
fetch: &F,
) -> Result<Vec<Candle>>
where
F: Fn(Option<Timestamp>, u32) -> Fut,
Fut: Future<Output = Result<Vec<Candle>>>,
G: Fn(Timestamp, i64) -> Option<Timestamp> + Copy,
{
if from >= upper {
return Ok(Vec::new());
}
let target = limit as usize;
let mut lower = from;
let mut calls = 0;
let mut collected = Vec::with_capacity(target);
while lower < upper && collected.len() < target {
let wanted = (target - collected.len()).min(max_per_call as usize) as u32;
let mut probe = CandleProbe {
exchange,
lower,
upper,
target,
collected: collected.len(),
fetch,
calls: &mut calls,
};
let chunk = oldest_chunk(&mut probe, wanted, max_per_call, advance).await?;
let complete = chunk.len() == wanted as usize;
let Some(last) = chunk.last().map(|candle| candle.open_time) else {
break;
};
collected.extend(chunk);
if !complete {
break;
}
let Some(next) = last.as_nanos().checked_add(1) else {
break;
};
lower = Timestamp::from_nanos(next);
}
collected.sort_by_key(|candle| candle.open_time);
collected.dedup_by_key(|candle| candle.open_time);
collected.truncate(target);
Ok(collected)
}
struct CandleProbe<'a, F> {
exchange: &'static str,
lower: Timestamp,
upper: Timestamp,
target: usize,
collected: usize,
fetch: &'a F,
calls: &'a mut u32,
}
async fn oldest_chunk<F, Fut, G>(
probe: &mut CandleProbe<'_, F>,
wanted: u32,
max_per_call: u32,
advance: G,
) -> Result<Vec<Candle>>
where
F: Fn(Option<Timestamp>, u32) -> Fut,
Fut: Future<Output = Result<Vec<Candle>>>,
G: Fn(Timestamp, i64) -> Option<Timestamp> + Copy,
{
let lower = probe.lower;
let upper = probe.upper;
let wanted_len = wanted as usize;
let mut low_steps = 0_i64;
let mut high_steps = i64::from(if wanted == max_per_call && wanted > 1 {
wanted - 1
} else {
wanted
});
let mut high_page;
if max_per_call == 1 {
let page = probe.run(lower, 1).await?;
if !page.is_empty() {
return Ok(page);
}
}
let request_count = wanted.saturating_add(1).min(max_per_call);
loop {
let end = bounded_advance(lower, high_steps, upper, advance)?;
high_page = probe.run(end, request_count).await?;
if high_page.len() >= wanted_len {
if low_steps == 0 {
high_page.truncate(wanted_len);
return Ok(high_page);
}
break;
}
if end == upper {
return Ok(high_page);
}
low_steps = high_steps;
high_steps = high_steps.checked_mul(2).unwrap_or(i64::MAX);
}
while high_steps - low_steps > 1 {
let middle = low_steps + (high_steps - low_steps) / 2;
let end = bounded_advance(lower, middle, upper, advance)?;
let page = probe.run(end, request_count).await?;
if page.len() >= wanted_len {
high_steps = middle;
high_page = page;
} else {
low_steps = middle;
}
}
high_page.truncate(wanted_len);
Ok(high_page)
}
fn bounded_advance<G>(
lower: Timestamp,
steps: i64,
upper: Timestamp,
advance: G,
) -> Result<Timestamp>
where
G: Fn(Timestamp, i64) -> Option<Timestamp>,
{
let end = advance(lower, steps).map_or(upper, |end| end.min(upper));
if end <= lower {
return Err(Error::invalid_request(
"interval",
"the provider candle grid did not advance",
));
}
Ok(end)
}
impl<F> CandleProbe<'_, F> {
async fn run<Fut>(&mut self, end: Timestamp, count: u32) -> Result<Vec<Candle>>
where
F: Fn(Option<Timestamp>, u32) -> Fut,
Fut: Future<Output = Result<Vec<Candle>>>,
{
if *self.calls >= MAX_CALLS {
return Err(Error::exchange(
self.exchange,
"candle_pagination_ceiling",
format!(
"walked {MAX_CALLS} pages of {} candles and gathered {} of the {} asked for",
self.exchange, self.collected, self.target
),
));
}
*self.calls += 1;
let mut page = (self.fetch)(Some(end), count).await?;
page.retain(|candle| {
candle.open_time >= self.lower
&& if end < self.upper {
candle.open_time <= end
} else {
candle.open_time < end
}
});
page.sort_by_key(|candle| candle.open_time);
page.dedup_by_key(|candle| candle.open_time);
Ok(page)
}
}
fn window_candles<G>(request: &CandleRequest, from: Timestamp, advance: G) -> usize
where
G: Fn(Timestamp, i64) -> Option<Timestamp>,
{
let end = request.to.unwrap_or_else(Timestamp::now);
let inside = |count: i64| advance(from, count).is_some_and(|at| at < end);
if !inside(0) {
return 0;
}
let mut outside: i64 = 1;
while inside(outside) {
let Some(doubled) = outside.checked_mul(2) else {
return usize::MAX;
};
outside = doubled;
}
let mut last_inside = outside / 2;
while outside - last_inside > 1 {
let midpoint = last_inside + (outside - last_inside) / 2;
if inside(midpoint) {
last_inside = midpoint;
} else {
outside = midpoint;
}
}
usize::try_from(last_inside + 1).unwrap_or(usize::MAX)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::types::{Exchange, Interval, Market};
use std::cell::RefCell;
const EXCHANGE: &str = "test";
const MINUTE_MS: i64 = 60_000;
fn market() -> Market {
Market::spot(Exchange::Upbit, "BTC", "KRW")
}
fn candle(open_ms: i64) -> Candle {
Candle {
market: market(),
interval: Interval::Min1,
open_time: Timestamp::from_millis(open_ms),
open: 1.into(),
high: 1.into(),
low: 1.into(),
close: 1.into(),
volume: 1.into(),
quote_volume: None,
closed: true,
}
}
struct Fake {
history: Vec<i64>,
inclusive_end: bool,
calls: RefCell<Vec<(Option<i64>, u32)>>,
}
impl Fake {
fn new(minutes: std::ops::Range<i64>) -> Self {
Self {
history: minutes.collect(),
inclusive_end: false,
calls: RefCell::new(Vec::new()),
}
}
fn inclusive(minutes: std::ops::Range<i64>) -> Self {
Self {
inclusive_end: true,
..Self::new(minutes)
}
}
fn page(&self, end: Option<Timestamp>, count: u32) -> Result<Vec<Candle>> {
self.calls
.borrow_mut()
.push((end.map(Timestamp::as_millis), count));
let end_ms = end.map_or(i64::MAX, Timestamp::as_millis);
let mut opens: Vec<i64> = self
.history
.iter()
.map(|minute| minute * MINUTE_MS)
.filter(|open| {
if self.inclusive_end {
*open <= end_ms
} else {
*open < end_ms
}
})
.collect();
opens.reverse();
opens.truncate(count as usize + usize::from(self.inclusive_end));
Ok(opens.into_iter().map(candle).collect())
}
}
async fn read_from(fake: &Fake, request: &CandleRequest, cap: u32) -> Result<Vec<i64>> {
let candles = read(request, EXCHANGE, cap, |end, count| async move {
fake.page(end, count)
})
.await?;
Ok(candles
.iter()
.map(|candle| candle.open_time.as_millis() / MINUTE_MS)
.collect())
}
fn request() -> CandleRequest {
CandleRequest::new(market(), Interval::Min1)
}
fn minute(minute: i64) -> Timestamp {
Timestamp::from_millis(minute * MINUTE_MS)
}
fn month(year: i32, month: u32) -> Timestamp {
let first = chrono::NaiveDate::from_ymd_opt(year, month, 1).expect("a first of the month");
Timestamp::from_secs(first.and_time(chrono::NaiveTime::MIN).and_utc().timestamp())
}
fn kst_month(year: i32, month_number: u32) -> Timestamp {
Timestamp::from_secs(month(year, month_number).as_secs() - 9 * 3_600)
}
fn advance_kst_month(at: Timestamp, count: i64) -> Option<Timestamp> {
let shifted = Timestamp::from_secs(at.as_secs().checked_add(9 * 3_600)?);
Interval::Month1
.advance(shifted, count)?
.as_secs()
.checked_sub(9 * 3_600)
.map(Timestamp::from_secs)
}
struct Monthly {
history: Vec<Timestamp>,
calls: RefCell<usize>,
}
impl Monthly {
fn new() -> Self {
Self {
history: (0..300)
.map(|n: i32| month(2000 + n / 12, u32::try_from(n % 12 + 1).expect("1..=12")))
.collect(),
calls: RefCell::new(0),
}
}
fn page(&self, end: Option<Timestamp>, count: u32) -> Result<Vec<Candle>> {
*self.calls.borrow_mut() += 1;
let end = end.unwrap_or(Timestamp::from_nanos(i64::MAX));
let mut opens: Vec<Timestamp> = self
.history
.iter()
.copied()
.filter(|at| *at < end)
.collect();
opens.reverse();
opens.truncate(count as usize);
Ok(opens
.into_iter()
.map(|open_time| Candle {
interval: Interval::Month1,
open_time,
..candle(0)
})
.collect())
}
}
async fn read_months(fake: &Monthly, request: &CandleRequest) -> Result<Vec<Timestamp>> {
let candles = read(request, EXCHANGE, 10, |end, count| async move {
fake.page(end, count)
})
.await?;
Ok(candles.iter().map(|candle| candle.open_time).collect())
}
struct ThirtyDay {
buckets: Vec<i64>,
calls: RefCell<usize>,
}
const THIRTY_DAYS_MS: i64 = 30 * 24 * 60 * 60 * 1_000;
fn bucket(index: i64) -> Timestamp {
Timestamp::from_millis(index * THIRTY_DAYS_MS)
}
impl ThirtyDay {
fn new(indexes: std::ops::Range<i64>) -> Self {
Self {
buckets: indexes.map(|index| index * THIRTY_DAYS_MS).collect(),
calls: RefCell::new(0),
}
}
fn page(&self, end: Option<Timestamp>, count: u32) -> Result<Vec<Candle>> {
*self.calls.borrow_mut() += 1;
let end_ms = end.map_or(i64::MAX, Timestamp::as_millis);
let mut opens: Vec<i64> = self
.buckets
.iter()
.copied()
.filter(|open| *open <= end_ms)
.collect();
opens.reverse();
opens.truncate(count as usize + 1);
Ok(opens
.into_iter()
.map(|open_ms| Candle {
interval: Interval::Month1,
open_time: Timestamp::from_millis(open_ms),
..candle(0)
})
.collect())
}
}
async fn read_thirty_day(fake: &ThirtyDay, request: &CandleRequest) -> Result<Vec<Timestamp>> {
let candles = read(request, EXCHANGE, 10, |end, count| async move {
fake.page(end, count)
})
.await?;
Ok(candles.iter().map(|candle| candle.open_time).collect())
}
#[tokio::test]
async fn a_window_on_a_grid_shorter_than_the_interval_answers_with_every_candle_it_holds() {
let fake = ThirtyDay::new(0..200);
let from = bucket(100);
let request = CandleRequest::new(market(), Interval::Month1)
.from(from)
.to(bucket(180));
let months = read_thirty_day(&fake, &request)
.await
.expect("a window inside the ceiling");
assert_eq!(
months.len(),
80,
"the window holds 80 buckets and the answer dropped some"
);
assert_eq!(months.first().copied(), Some(from));
assert_eq!(months.last().copied(), Some(bucket(179)));
}
#[tokio::test]
async fn a_start_time_and_a_count_on_a_shorter_grid_answer_the_oldest_not_the_newest() {
let fake = ThirtyDay::new(0..200);
let request = CandleRequest::new(market(), Interval::Month1)
.from(bucket(100))
.limit(12);
let months = read_thirty_day(&fake, &request).await.expect("ok");
assert_eq!(months.len(), 12);
assert_eq!(
months,
(100..112).map(bucket).collect::<Vec<_>>(),
"the oldest twelve at or after `from`, not the newest twelve before the cursor"
);
}
#[tokio::test]
async fn a_limit_within_one_page_costs_one_call_and_returns_the_newest_candles() {
let fake = Fake::new(0..100);
assert_eq!(
read_from(&fake, &request().limit(3), 10).await.expect("ok"),
vec![97, 98, 99]
);
assert_eq!(fake.calls.borrow().len(), 1);
}
#[tokio::test]
async fn a_limit_above_the_cap_is_paged_rather_than_refused() {
let fake = Fake::new(0..100);
assert_eq!(
read_from(&fake, &request().limit(25), 10)
.await
.expect("ok"),
(75..100).collect::<Vec<_>>()
);
assert_eq!(fake.calls.borrow().len(), 3);
}
#[tokio::test]
async fn a_start_time_is_honoured_by_walking_back_to_it() {
let latest = Timestamp::now().as_secs() / 60;
let fake = Fake::new(latest - 100..latest);
assert_eq!(
read_from(&fake, &request().from(minute(latest - 40)), 10)
.await
.expect("ok"),
(latest - 40..latest).collect::<Vec<_>>()
);
}
#[tokio::test]
async fn a_start_time_with_a_count_answers_from_the_start_not_from_the_present() {
let fake = Fake::new(0..100);
assert_eq!(
read_from(&fake, &request().from(minute(40)).limit(5), 10)
.await
.expect("ok"),
vec![40, 41, 42, 43, 44]
);
assert_eq!(fake.calls.borrow().len(), 1);
}
#[tokio::test]
async fn a_start_time_with_a_count_skips_missing_candle_windows() {
let fake = Fake {
history: vec![0, 2],
inclusive_end: false,
calls: RefCell::new(Vec::new()),
};
assert_eq!(
read_from(&fake, &request().from(minute(0)).limit(2), 10)
.await
.expect("two candles across an empty window"),
vec![0, 2]
);
}
#[tokio::test]
async fn an_inclusive_cursor_can_use_one_of_the_requested_slots() {
let request = request().from(minute(0)).to(minute(2)).limit(2);
let candles = read(&request, EXCHANGE, 2, |end, count| async move {
let end = end.expect("a bounded probe").as_millis() / MINUTE_MS;
let mut opens = [0, 1, 2]
.into_iter()
.filter(|open| *open <= end)
.collect::<Vec<_>>();
opens.reverse();
opens.truncate(count as usize);
Ok(opens
.into_iter()
.map(|open| candle(open * MINUTE_MS))
.collect())
})
.await
.expect("an inclusive provider page");
assert_eq!(
candles
.into_iter()
.map(|candle| candle.open_time.as_millis() / MINUTE_MS)
.collect::<Vec<_>>(),
vec![0, 1]
);
}
#[tokio::test]
async fn a_long_empty_span_is_searched_without_one_call_per_window() {
let fake = Fake {
history: vec![0, 10_000, 10_001],
inclusive_end: false,
calls: RefCell::new(Vec::new()),
};
assert_eq!(
read_from(&fake, &request().from(minute(0)).limit(3), 10)
.await
.expect("three candles across a long empty span"),
vec![0, 10_000, 10_001]
);
assert!(
fake.calls.borrow().len() < 40,
"the search should expand and bisect the empty span"
);
}
#[tokio::test]
async fn an_end_time_can_leave_a_from_limit_request_short() {
let fake = Fake {
history: vec![0, 2, 4],
inclusive_end: false,
calls: RefCell::new(Vec::new()),
};
assert_eq!(
read_from(&fake, &request().from(minute(0)).to(minute(3)).limit(3), 10,)
.await
.expect("the bounded history"),
vec![0, 2]
);
}
#[tokio::test]
async fn an_end_time_is_exclusive_and_a_start_time_inclusive() {
let fake = Fake::new(0..100);
assert_eq!(
read_from(&fake, &request().from(minute(40)).to(minute(43)), 10)
.await
.expect("ok"),
vec![40, 41, 42]
);
}
#[tokio::test]
async fn a_history_shorter_than_the_request_ends_the_walk() {
let fake = Fake::new(95..100);
assert_eq!(
read_from(&fake, &request().limit(50), 10)
.await
.expect("ok"),
(95..100).collect::<Vec<_>>()
);
assert_eq!(fake.calls.borrow().len(), 1);
}
#[tokio::test]
async fn a_cursor_that_stops_moving_is_reported_instead_of_looping() {
let calls = RefCell::new(0u32);
let stuck = read(&request().limit(30), EXCHANGE, 10, |_, _| {
*calls.borrow_mut() += 1;
async { Ok((0..10).map(|minute| candle(minute * MINUTE_MS)).collect()) }
})
.await;
assert!(matches!(stuck, Err(Error::Exchange { .. })));
assert!(*calls.borrow() < 5, "the walk did not stop");
}
#[tokio::test]
async fn a_start_time_with_no_count_is_bounded_before_the_first_call() {
let fake = Fake::new(0..100);
assert!(matches!(
read_from(&fake, &request().from(Timestamp::from_secs(0)), 10).await,
Err(Error::InvalidRequest { field, .. }) if field == "from"
));
assert!(
fake.calls.borrow().is_empty(),
"a window that cannot be walked should not be started"
);
}
#[tokio::test]
async fn a_count_past_what_the_walk_can_reach_is_refused_with_the_number() {
let fake = Fake::new(0..100);
let ceiling = 10 * MAX_CALLS as usize;
assert!(
read_from(&fake, &request().limit(ceiling as u32), 10)
.await
.is_ok()
);
let Err(Error::InvalidRequest { field, detail }) =
read_from(&fake, &request().limit(ceiling as u32 + 1), 10).await
else {
panic!("a count past the ceiling should be refused");
};
assert_eq!(field, "limit");
assert!(
detail.contains(&ceiling.to_string()),
"the refusal should say how much can be read: {detail}"
);
}
#[tokio::test]
async fn the_first_candle_a_market_ever_had_ends_the_walk_rather_than_failing_it() {
let fake = Fake::inclusive(0..10);
assert_eq!(
read_from(&fake, &request().limit(25), 10)
.await
.expect("the end of a history is not an error"),
(0..10).collect::<Vec<_>>()
);
assert_eq!(fake.calls.borrow().len(), 2);
}
#[tokio::test]
async fn a_start_time_with_a_count_answers_from_the_start_at_monthly_candles_too() {
let fake = Monthly::new();
let request = CandleRequest::new(market(), Interval::Month1)
.from(month(2000, 5))
.limit(5);
assert_eq!(
read_months(&fake, &request).await.expect("ok"),
vec![
month(2000, 5),
month(2000, 6),
month(2000, 7),
month(2000, 8),
month(2000, 9)
]
);
assert_eq!(*fake.calls.borrow(), 1, "the window is known before asking");
}
#[tokio::test]
async fn a_provider_calendar_grid_sets_the_from_limit_window() {
let history = [kst_month(2026, 3), kst_month(2026, 4)];
let request = CandleRequest::new(market(), Interval::Month1)
.from(Timestamp::from_secs(history[0].as_secs() + 1))
.limit(1);
let candles = read_on_grid(
&request,
EXCHANGE,
10,
advance_kst_month,
|end: Option<Timestamp>, count: u32| {
let history = &history;
async move {
let end = end.unwrap_or(Timestamp::from_nanos(i64::MAX));
let mut opens: Vec<_> =
history.iter().copied().filter(|open| *open < end).collect();
opens.reverse();
opens.truncate(count as usize);
Ok(opens
.into_iter()
.map(|open_time| Candle {
interval: Interval::Month1,
open_time,
..candle(0)
})
.collect())
}
},
)
.await
.expect("a KST monthly window");
assert_eq!(
candles
.into_iter()
.map(|candle| candle.open_time)
.collect::<Vec<_>>(),
vec![history[1]]
);
}
#[tokio::test]
async fn a_monthly_window_at_the_ceiling_is_measured_in_real_months() {
let fake = Monthly::new();
let request = CandleRequest::new(market(), Interval::Month1)
.from(month(1950, 1))
.to(month(2033, 5));
let months = read_months(&fake, &request)
.await
.expect("a thousand monthly candles is inside the ceiling");
assert_eq!(months.len(), 300);
assert_eq!(months.first().copied(), Some(month(2000, 1)));
assert_eq!(months.last().copied(), Some(month(2024, 12)));
}
#[tokio::test]
async fn a_page_that_re_serves_the_candle_at_the_cursor_is_not_a_stall() {
let fake = Fake::inclusive(0..100_000);
assert_eq!(
read_from(&fake, &request().limit(25), 10)
.await
.expect("a mid-history walk is not a stalled cursor"),
(99_975..100_000).collect::<Vec<_>>()
);
}
#[tokio::test]
async fn no_request_costs_more_calls_than_the_ceiling() {
let calls = RefCell::new(0u32);
let walk = read(
&request().limit(10 * MAX_CALLS),
EXCHANGE,
10,
|end, count| {
*calls.borrow_mut() += 1;
let newest = end.map_or(1_000_000, |end| end.as_millis() / MINUTE_MS);
async move {
Ok((0..i64::from(count))
.map(|back| candle((newest - back) * MINUTE_MS))
.collect())
}
},
)
.await;
assert!(matches!(walk, Err(Error::Exchange { .. })), "{walk:?}");
assert_eq!(*calls.borrow(), MAX_CALLS);
}
#[tokio::test]
async fn a_count_alone_answers_the_newest_even_when_a_page_carries_one_extra() {
let fake = Fake::inclusive(0..100);
assert_eq!(
read_from(&fake, &request().limit(1), 10).await.expect("ok"),
vec![99],
"one candle asked for with no start time is the newest one"
);
}
#[tokio::test]
async fn a_window_running_past_the_end_of_time_is_refused_not_answered_from_the_present() {
let fake = Fake::new(0..100);
let request = request().from(Timestamp::from_nanos(i64::MAX - 1)).limit(2);
assert!(
matches!(
read_from(&fake, &request, 10).await,
Err(Error::InvalidRequest { field, .. }) if field == "from"
),
"a window that cannot be expressed should not be answered"
);
assert!(fake.calls.borrow().is_empty());
}
#[tokio::test]
async fn an_explicit_end_can_bound_a_limit_that_would_otherwise_overflow() {
let fake = Fake::new(0..100);
let from = Timestamp::from_nanos(i64::MAX - 120_000_000_000);
let to = Timestamp::from_nanos(i64::MAX - 60_000_000_000);
let request = request().from(from).to(to).limit(10);
assert!(
read_from(&fake, &request, 10).await.is_ok(),
"the explicit end keeps the requested window representable"
);
}
#[tokio::test]
async fn a_request_for_no_candles_is_a_caller_mistake() {
let fake = Fake::new(0..100);
assert!(matches!(
read_from(&fake, &request().limit(0), 10).await,
Err(Error::InvalidRequest { field, .. }) if field == "limit"
));
assert!(fake.calls.borrow().is_empty());
}
#[tokio::test]
async fn a_window_that_ends_before_it_starts_never_reaches_the_wire() {
let fake = Fake::new(0..100);
assert!(matches!(
read_from(&fake, &request().from(minute(50)).to(minute(50)), 10).await,
Err(Error::InvalidRequest { field, .. }) if field == "from"
));
assert!(fake.calls.borrow().is_empty());
}
#[tokio::test]
async fn an_over_cap_count_of_monthly_candles_with_no_start_time_pages_like_any_other() {
let fake = Monthly::new();
let over_cap = CandleRequest::new(market(), Interval::Month1).limit(25);
let months = read_months(&fake, &over_cap)
.await
.expect("an unanchored monthly count pages");
assert_eq!(months.len(), 25);
assert_eq!(months.last().copied(), Some(month(2024, 12)));
assert_eq!(months.first().copied(), Some(month(2022, 12)));
assert_eq!(*fake.calls.borrow(), 3);
}
}