refactor and test epoch scheduling

contracts/consumer/remote-price-feed/src/contract.rs

@@ -1,4 +1,4 @@
-use cosmwasm_std::{entry_point, DepsMut, Env, IbcChannel, Response, Timestamp};
+use cosmwasm_std::{entry_point, DepsMut, Env, IbcChannel, Response};
 use cw2::set_contract_version;
 use cw_storage_plus::Item;
 use cw_utils::nonpayable;
@@ -11,6 +11,7 @@ use mesh_apis::price_feed_api::{self, PriceFeedApi, PriceResponse};
 use crate::error::ContractError;
 use crate::ibc::{make_ibc_packet, AUTH_ENDPOINT};
 use crate::msg::AuthorizedEndpoint;
+use crate::scheduler::{Action, Scheduler};
 use crate::state::{PriceInfo, TradingPair};
 
 pub const CONTRACT_NAME: &str = env!("CARGO_PKG_NAME");
@@ -20,31 +21,31 @@ pub struct RemotePriceFeedContract {
     pub channel: Item<'static, IbcChannel>,
     pub trading_pair: Item<'static, TradingPair>,
     pub price_info: Item<'static, PriceInfo>,
-    pub last_epoch: Item<'static, Timestamp>,
-    pub epoch_in_secs: Item<'static, u64>,
     pub price_info_ttl_in_secs: Item<'static, u64>,
+    pub scheduler: Scheduler<Box<dyn Action>>,
 }
 
 #[cfg_attr(not(feature = "library"), sylvia::entry_points)]
 #[contract]
 #[error(ContractError)]
 #[messages(price_feed_api as PriceFeedApi)]
 impl RemotePriceFeedContract {
-    pub const fn new() -> Self {
+    pub fn new() -> Self {
         Self {
             channel: Item::new("channel"),
             trading_pair: Item::new("tpair"),
             price_info: Item::new("price"),
-            last_epoch: Item::new("last_epoch"),
-            epoch_in_secs: Item::new("epoch"),
             price_info_ttl_in_secs: Item::new("price_ttl"),
+            // TODO: the indirection can be removed once Sylvia supports
+            // generics
+            scheduler: Scheduler::new(Box::new(handle_epoch)),
         }
     }
 
     #[msg(instantiate)]
     pub fn instantiate(
         &self,
-        ctx: InstantiateCtx,
+        mut ctx: InstantiateCtx,
         trading_pair: TradingPair,
         auth_endpoint: AuthorizedEndpoint,
         epoch_in_secs: u64,
@@ -53,13 +54,12 @@ impl RemotePriceFeedContract {
         nonpayable(&ctx.info)?;
 
         set_contract_version(ctx.deps.storage, CONTRACT_NAME, CONTRACT_VERSION)?;
-        self.last_epoch
-            .save(ctx.deps.storage, &Timestamp::from_seconds(0))?;
         self.trading_pair.save(ctx.deps.storage, &trading_pair)?;
-        self.epoch_in_secs.save(ctx.deps.storage, &epoch_in_secs)?;
         self.price_info_ttl_in_secs
             .save(ctx.deps.storage, &price_info_ttl_in_secs)?;
 
+        self.scheduler.init(&mut ctx.deps, epoch_in_secs)?;
+
         AUTH_ENDPOINT.save(ctx.deps.storage, &auth_endpoint)?;
 
         Ok(Response::new())
@@ -93,36 +93,32 @@ impl PriceFeedApi for RemotePriceFeedContract {
 
 #[cfg_attr(not(feature = "library"), entry_point)]
 pub fn sudo(deps: DepsMut, env: Env, msg: SudoMsg) -> Result<Response, ContractError> {
+    let contract = RemotePriceFeedContract::new();
+
     match msg {
-        SudoMsg::HandleEpoch {} => handle_epoch(deps, env),
+        SudoMsg::HandleEpoch {} => contract.scheduler.trigger(deps, &env),
     }
 }
 
-pub fn handle_epoch(deps: DepsMut, env: Env) -> Result<Response, ContractError> {
+pub fn handle_epoch(deps: DepsMut, env: &Env) -> Result<Response, ContractError> {
     let contract = RemotePriceFeedContract::new();
     let TradingPair {
         pool_id,
         base_asset,
         quote_asset,
     } = contract.trading_pair.load(deps.storage)?;
-    let last_epoch = contract.last_epoch.load(deps.storage)?;
-    let epoch_duration = contract.epoch_in_secs.load(deps.storage)?;
-    let secs_since_last_epoch = env.block.time.seconds() - last_epoch.seconds();
-    if secs_since_last_epoch >= epoch_duration {
-        let channel = contract
-            .channel
-            .may_load(deps.storage)?
-            .ok_or(ContractError::IbcChannelNotOpen)?;
-
-        let packet = mesh_apis::ibc::RemotePriceFeedPacket::QueryTwap {
-            pool_id,
-            base_asset,
-            quote_asset,
-        };
-        let msg = make_ibc_packet(&env.block.time, channel, packet)?;
-
-        Ok(Response::new().add_message(msg))
-    } else {
-        Ok(Response::new())
-    }
+
+    let channel = contract
+        .channel
+        .may_load(deps.storage)?
+        .ok_or(ContractError::IbcChannelNotOpen)?;
+
+    let packet = mesh_apis::ibc::RemotePriceFeedPacket::QueryTwap {
+        pool_id,
+        base_asset,
+        quote_asset,
+    };
+    let msg = make_ibc_packet(&env.block.time, channel, packet)?;
+
+    Ok(Response::new().add_message(msg))
 }

contracts/consumer/remote-price-feed/src/lib.rs

@@ -2,4 +2,5 @@ pub mod contract;
 pub mod error;
 pub mod ibc;
 pub mod msg;
+pub mod scheduler;
 pub mod state;

contracts/consumer/remote-price-feed/src/scheduler.rs

@@ -0,0 +1,100 @@
+use cosmwasm_std::{DepsMut, Env, Response, Timestamp};
+use cw_storage_plus::Item;
+
+use crate::error::ContractError;
+
+pub trait Action: Fn(DepsMut, &Env) -> Result<Response, ContractError> {}
+impl<F> Action for F where F: Fn(DepsMut, &Env) -> Result<Response, ContractError> {}
+
+/// A helper to schedule a single action to be executed regularly,
+/// as in "every epoch". It relies on a trigger being called rather rapidly (every block?).
+pub struct Scheduler<A> {
+    last_epoch: Item<'static, Timestamp>,
+    epoch_in_secs: Item<'static, u64>,
+    action: A,
+}
+
+impl<A> Scheduler<A>
+where
+    A: Action,
+{
+    pub const fn new(action: A) -> Self {
+        Self {
+            last_epoch: Item::new("last_epoch"),
+            epoch_in_secs: Item::new("epoch"),
+            action,
+        }
+    }
+
+    pub fn init(&self, deps: &mut DepsMut, epoch_in_secs: u64) -> Result<(), ContractError> {
+        self.last_epoch
+            .save(deps.storage, &Timestamp::from_seconds(0))?;
+        self.epoch_in_secs.save(deps.storage, &epoch_in_secs)?;
+        Ok(())
+    }
+
+    pub fn trigger(&self, deps: DepsMut, env: &Env) -> Result<Response, ContractError> {
+        let last_epoch = self.last_epoch.load(deps.storage)?;
+        let epoch_in_secs = self.epoch_in_secs.load(deps.storage)?;
+        let secs_since_last_epoch = env.block.time.seconds() - last_epoch.seconds();
+        if secs_since_last_epoch >= epoch_in_secs {
+            self.last_epoch.save(deps.storage, &env.block.time)?;
+            (self.action)(deps, env)
+        } else {
+            Ok(Response::new())
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use cosmwasm_std::{
+        testing::{mock_dependencies, mock_env},
+        Binary,
+    };
+
+    #[test]
+    fn scheduler_first_epoch_always_fires() {
+        let scheduler = Scheduler::new(|_, _| Ok(Response::new().set_data(Binary::from(b"foo"))));
+        let mut deps = mock_dependencies();
+        let env = mock_env();
+
+        scheduler.init(&mut deps.as_mut(), 111111).unwrap();
+        assert!(scheduler
+            .trigger(deps.as_mut(), &env)
+            .unwrap()
+            .data
+            .is_some());
+    }
+
+    #[test]
+    fn scheduler() {
+        let scheduler = Scheduler::new(|_, _| Ok(Response::new().set_data(Binary::from(b"foo"))));
+        let mut deps = mock_dependencies();
+        let mut env = mock_env();
+
+        scheduler.init(&mut deps.as_mut(), 10).unwrap();
+
+        #[track_caller]
+        fn assert_fired<A: Action>(s: &Scheduler<A>, deps: DepsMut, env: &Env) {
+            assert!(s.trigger(deps, env).unwrap().data.is_some())
+        }
+
+        #[track_caller]
+        fn assert_noop<A: Action>(s: &Scheduler<A>, deps: DepsMut, env: &Env) {
+            assert!(s.trigger(deps, env).unwrap().data.is_none())
+        }
+
+        assert_fired(&scheduler, deps.as_mut(), &env);
+
+        env.block.time = env.block.time.plus_seconds(5);
+        assert_noop(&scheduler, deps.as_mut(), &env);
+        env.block.time = env.block.time.plus_seconds(5);
+        assert_fired(&scheduler, deps.as_mut(), &env);
+        env.block.time = env.block.time.plus_seconds(5);
+        assert_noop(&scheduler, deps.as_mut(), &env);
+        env.block.time = env.block.time.plus_seconds(5);
+        assert_fired(&scheduler, deps.as_mut(), &env);
+    }
+}