From e42df1a058c1f4a1c9894b97446842da36c54ebd Mon Sep 17 00:00:00 2001
From: Nathaniel Cook <nvcook42@gmail.com>
Date: Wed, 16 Oct 2024 15:25:37 -0600
Subject: [PATCH] fix: use tokio::time::Interval for scheduling anchor batches
 (#563)

* 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.

* fix: use division operator

* fix: better positive check logic

* use function and add comment

* fixup
---
 anchor-service/src/anchor_batch.rs | 67 ++++++++++++++++++++----------
 1 file changed, 46 insertions(+), 21 deletions(-)

diff --git a/anchor-service/src/anchor_batch.rs b/anchor-service/src/anchor_batch.rs
index f3614934..9ec9ab7d 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, Interval, MissedTickBehavior};
 use tracing::{error, info};
 
 use crate::high_water_mark_store::HighWaterMarkStore;
@@ -76,28 +77,15 @@ impl AnchorService {
         pin_mut!(shutdown_signal);
 
         info!("anchor service started");
+        let mut interval = self.build_interval();
 
         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((_, _)) => {
@@ -108,6 +96,43 @@ 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");
+        // 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.
+        // 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)
+            .expect("truncated duration should be in range")
+            + period
+            - buffer;
+
+        // Ensure delay is always positive, in order to construct [`std::time::Duration`].
+        let delay = if next_tick > now {
+            next_tick - now
+        } else {
+            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);
+        interval
+    }
+
     async fn process_next_batch(&mut self) -> Result<()> {
         // Pass the anchor requests through a deduplication step to avoid anchoring multiple Data Events from the same
         // Stream.