From 75f815b49e860544fc67c9128e2b7b7392a4dbb1 Mon Sep 17 00:00:00 2001 From: Nathaniel Cook Date: Wed, 16 Oct 2024 13:30:17 -0600 Subject: [PATCH 1/5] fix: use tokio::time::Interval for scheduling anchor batches This leverages the time math of skipping missed ticks etc. We are only left with computing the time when we should start the first tick. For this we use chrono and then rely on Interval for the heavy lifting. --- anchor-service/src/anchor_batch.rs | 57 +++++++++++++++++++----------- 1 file changed, 36 insertions(+), 21 deletions(-) diff --git a/anchor-service/src/anchor_batch.rs b/anchor-service/src/anchor_batch.rs index f3614934..d385a300 100644 --- a/anchor-service/src/anchor_batch.rs +++ b/anchor-service/src/anchor_batch.rs @@ -3,12 +3,13 @@ use async_trait::async_trait; use ceramic_core::{Cid, NodeId}; use ceramic_sql::sqlite::SqlitePool; use chrono::DurationRound; -use chrono::{Duration as ChronoDuration, TimeDelta, Utc}; +use chrono::{TimeDelta, Utc}; use futures::future::{select, Either, FutureExt}; use futures::pin_mut; use indexmap::IndexMap; use std::future::Future; use std::{sync::Arc, time::Duration}; +use tokio::time::{interval_at, Instant, MissedTickBehavior}; use tracing::{error, info}; use crate::high_water_mark_store::HighWaterMarkStore; @@ -76,28 +77,42 @@ impl AnchorService { pin_mut!(shutdown_signal); info!("anchor service started"); + let period = + TimeDelta::from_std(self.anchor_interval).expect("anchor interval should be in range"); + let buffer = TimeDelta::from_std(self.anchor_interval / 12) + .expect("anchor interval fraction should be in range"); + + // It is not possible to truncate a tokio::time::Instant as the time is opaque. + // Therefore we use the `chrono` crate to do the truncate logic and then compute the delay + // from now till the next tick. This way we can then simply add the delay to the + // Instant::now() to get the equivalent instant. + // Get both the chrono and Instant now times close together. + let instant_now = Instant::now(); + let now = Utc::now(); + let next_tick = now.duration_trunc(period).unwrap() + period - buffer; + + let delay = next_tick - now; + // Durations in rust are always positive. + // If the delay is negative, it means the next tick is in the past, therefore we should set + // the delay to be plus one period. + let delay = if let Ok(delay) = delay.to_std() { + delay + } else { + (delay + period) + .to_std() + .expect("delay should be positive as its should be in the next interval") + }; + // Start an interval at the next tick instant. + let mut interval = interval_at(instant_now + delay, self.anchor_interval); + interval.set_missed_tick_behavior(MissedTickBehavior::Skip); loop { - let now = Utc::now(); - let next_tick = now - .duration_trunc(TimeDelta::from_std(self.anchor_interval).unwrap()) - .unwrap() - + TimeDelta::from_std(self.anchor_interval).unwrap() - - ChronoDuration::minutes(5); - - let delay = next_tick - now; - // durations in rust are always positive. - // If the delay is negative, it means the next tick is in the past, therefore - // we should process the next batch immediately. - if let Ok(delay) = delay.to_std() { - tokio::time::sleep(delay).await; - } - let process_next_batch = self.process_next_batch(); - pin_mut!(process_next_batch); - match select(process_next_batch, &mut shutdown_signal).await { - Either::Left((result, _)) => { - if let Err(e) = result { - error!("Error processing batch: {:?}", e); + let tick = interval.tick(); + pin_mut!(tick); + match select(tick, &mut shutdown_signal).await { + Either::Left(_) => { + if let Err(err) = self.process_next_batch().await { + error!(%err, "error processing batch"); } } Either::Right((_, _)) => { From 5a2ed625262f0989415131639a6bcf3c1eda22a9 Mon Sep 17 00:00:00 2001 From: Nathaniel Cook Date: Wed, 16 Oct 2024 13:35:09 -0600 Subject: [PATCH 2/5] fix: use division operator --- anchor-service/src/anchor_batch.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/anchor-service/src/anchor_batch.rs b/anchor-service/src/anchor_batch.rs index d385a300..ebc2c48d 100644 --- a/anchor-service/src/anchor_batch.rs +++ b/anchor-service/src/anchor_batch.rs @@ -79,8 +79,7 @@ impl AnchorService { info!("anchor service started"); let period = TimeDelta::from_std(self.anchor_interval).expect("anchor interval should be in range"); - let buffer = TimeDelta::from_std(self.anchor_interval / 12) - .expect("anchor interval fraction should be in range"); + let buffer = period / 12; // It is not possible to truncate a tokio::time::Instant as the time is opaque. // Therefore we use the `chrono` crate to do the truncate logic and then compute the delay From d7210efde9b3a37e353eaf2282020a7cb28af14e Mon Sep 17 00:00:00 2001 From: Nathaniel Cook Date: Wed, 16 Oct 2024 13:49:02 -0600 Subject: [PATCH 3/5] fix: better positive check logic --- anchor-service/src/anchor_batch.rs | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/anchor-service/src/anchor_batch.rs b/anchor-service/src/anchor_batch.rs index ebc2c48d..e9bab717 100644 --- a/anchor-service/src/anchor_batch.rs +++ b/anchor-service/src/anchor_batch.rs @@ -88,19 +88,20 @@ impl AnchorService { // Get both the chrono and Instant now times close together. let instant_now = Instant::now(); let now = Utc::now(); - let next_tick = now.duration_trunc(period).unwrap() + period - buffer; + let next_tick = now + .duration_trunc(period) + .expect("truncated duration should be in range") + + period + - buffer; - let delay = next_tick - now; - // Durations in rust are always positive. - // If the delay is negative, it means the next tick is in the past, therefore we should set - // the delay to be plus one period. - let delay = if let Ok(delay) = delay.to_std() { - delay + // Ensure delay is always positive, in order to construct [`std::time::Duration`]. + let delay = if next_tick > now { + next_tick - now } else { - (delay + period) - .to_std() - .expect("delay should be positive as its should be in the next interval") + next_tick - now + period }; + let delay = delay.to_std().expect("delay should always be positive"); + // Start an interval at the next tick instant. let mut interval = interval_at(instant_now + delay, self.anchor_interval); interval.set_missed_tick_behavior(MissedTickBehavior::Skip); From aeaafabba61f045f81d84d980cbfe7eda3e1d77c Mon Sep 17 00:00:00 2001 From: Nathaniel Cook Date: Wed, 16 Oct 2024 14:18:51 -0600 Subject: [PATCH 4/5] use function and add comment --- anchor-service/src/anchor_batch.rs | 42 ++++++++++++++++++------------ 1 file changed, 25 insertions(+), 17 deletions(-) diff --git a/anchor-service/src/anchor_batch.rs b/anchor-service/src/anchor_batch.rs index e9bab717..3744139d 100644 --- a/anchor-service/src/anchor_batch.rs +++ b/anchor-service/src/anchor_batch.rs @@ -9,7 +9,7 @@ use futures::pin_mut; use indexmap::IndexMap; use std::future::Future; use std::{sync::Arc, time::Duration}; -use tokio::time::{interval_at, Instant, MissedTickBehavior}; +use tokio::time::{interval_at, Instant, Interval, MissedTickBehavior}; use tracing::{error, info}; use crate::high_water_mark_store::HighWaterMarkStore; @@ -77,8 +77,31 @@ impl AnchorService { pin_mut!(shutdown_signal); info!("anchor service started"); + let interval = self.build_interval(); + + loop { + let tick = interval.tick(); + pin_mut!(tick); + match select(tick, &mut shutdown_signal).await { + Either::Left(_) => { + if let Err(err) = self.process_next_batch().await { + error!(%err, "error processing batch"); + } + } + Either::Right((_, _)) => { + break; + } + } + } + info!("anchor service stopped"); + } + + fn build_interval(&self) -> Interval { let period = TimeDelta::from_std(self.anchor_interval).expect("anchor interval should be in range"); + // We want the interval to tick on the period with a buffer time before the period begins. + // Buffer is somewhat arbitrarily defined as 1/12th the period. For a period of one hour + // the buffer is 5 minutes, meaning we tick 5 minutes before each hour. let buffer = period / 12; // It is not possible to truncate a tokio::time::Instant as the time is opaque. @@ -105,22 +128,7 @@ impl AnchorService { // Start an interval at the next tick instant. let mut interval = interval_at(instant_now + delay, self.anchor_interval); interval.set_missed_tick_behavior(MissedTickBehavior::Skip); - - loop { - let tick = interval.tick(); - pin_mut!(tick); - match select(tick, &mut shutdown_signal).await { - Either::Left(_) => { - if let Err(err) = self.process_next_batch().await { - error!(%err, "error processing batch"); - } - } - Either::Right((_, _)) => { - break; - } - } - } - info!("anchor service stopped"); + interval } async fn process_next_batch(&mut self) -> Result<()> { From 4eeaa489aed17b20552903b622546eb52b9df02c Mon Sep 17 00:00:00 2001 From: Nathaniel Cook Date: Wed, 16 Oct 2024 15:00:48 -0600 Subject: [PATCH 5/5] fixup --- anchor-service/src/anchor_batch.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/anchor-service/src/anchor_batch.rs b/anchor-service/src/anchor_batch.rs index 3744139d..9ec9ab7d 100644 --- a/anchor-service/src/anchor_batch.rs +++ b/anchor-service/src/anchor_batch.rs @@ -77,7 +77,7 @@ impl AnchorService { pin_mut!(shutdown_signal); info!("anchor service started"); - let interval = self.build_interval(); + let mut interval = self.build_interval(); loop { let tick = interval.tick(); @@ -96,6 +96,8 @@ impl AnchorService { info!("anchor service stopped"); } + // Construct an [`Interval`] that will tick just before each anchor interval period. + // Missed ticks will skip. fn build_interval(&self) -> Interval { let period = TimeDelta::from_std(self.anchor_interval).expect("anchor interval should be in range");