Scheduled reducer: use timestamp from reducer params for next run (#3657)

# Description of Changes

fixes: #2882

Schedule repeated reducers from their last execution timestamp instead
of `Timestamp::now`

# API and ABI breaking changes
NA

# Expected complexity level and risk
1

# Testing
Existing test should be enough to cover any regression.

Author: Shubham8287

crates/core/src/host/module_host.rs

@@ -1491,7 +1491,7 @@ impl ModuleHost {
     pub async fn call_scheduled_reducer(
         &self,
         call_reducer_params: impl FnOnce(&MutTxId) -> anyhow::Result<Option<CallReducerParams>> + Send + 'static,
-    ) -> Result<ReducerCallResult, ReducerCallError> {
+    ) -> Result<(ReducerCallResult, Timestamp), ReducerCallError> {
         let db = self.module.replica_ctx().relational_db.clone();
         // scheduled reducer name not fetched yet, anyway this is only for logging purpose
         const REDUCER: &str = "scheduled_reducer";
@@ -1518,8 +1518,9 @@ impl ModuleHost {
                             arg_bsatn: params.args.get_bsatn().clone(),
                         }),
                     );
+                    let timestamp = params.timestamp;
 
-                    Ok(inst.call_reducer(Some(tx), params))
+                    Ok((inst.call_reducer(Some(tx), params), timestamp))
                 }
                 Ok(None) => Err(ReducerCallError::ScheduleReducerNotFound),
                 Err(err) => Err(ReducerCallError::Args(InvalidReducerArguments(

crates/core/src/host/scheduler.rs

@@ -396,11 +396,19 @@ impl SchedulerActor {
             // the ScheduledReducer in the database for when the module restarts
             Ok(Err(ReducerCallError::NoSuchModule(_)) | Err(ReducerCallError::ScheduleReducerNotFound)) => {}
 
+            Ok(Ok((_, ts))) => {
+                if let Some(id) = id {
+                    let _ = self.delete_scheduled_reducer_row(&db, id, module_host_clone, ts).await;
+                }
+            }
+
             // delete the scheduled reducer row if its not repeated reducer
             Ok(_) | Err(_) => {
                 if let Some(id) = id {
                     // TODO: Handle errors here?
-                    let _ = self.delete_scheduled_reducer_row(&db, id, module_host_clone).await;
+                    let _ = self
+                        .delete_scheduled_reducer_row(&db, id, module_host_clone, Timestamp::now())
+                        .await;
                 }
             }
         }
@@ -415,6 +423,7 @@ impl SchedulerActor {
         db: &RelationalDB,
         id: ScheduledReducerId,
         module_host: ModuleHost,
+        ts: Timestamp,
     ) -> anyhow::Result<()> {
         let host_clone = module_host.clone();
         let db = db.clone();
@@ -455,10 +464,7 @@ impl SchedulerActor {
         // If this was repeated, we need to add it back to the queue.
         if let Some(ScheduleAt::Interval(dur)) = schedule_at {
             let key = self.queue.insert(
-                QueueItem::Id {
-                    id,
-                    at: Timestamp::now() + dur,
-                },
+                QueueItem::Id { id, at: ts + dur },
                 dur.to_duration().unwrap_or(Duration::ZERO),
             );
             self.key_map.insert(id, key);