Rounding time to step in flow system to get stable activation times

Author: zaychenko-sergei

src/app/cli/src/app.rs

@@ -266,6 +266,9 @@ pub fn configure_base_catalog(
 
     b.add::<kamu_flow_system_inmem::FlowConfigurationServiceInMemory>();
     b.add::<kamu_flow_system_inmem::FlowServiceInMemory>();
+    b.add_value(kamu_flow_system_inmem::domain::FlowServiceRunConfig::new(
+        chrono::Duration::seconds(1),
+    ));
 
     b.add::<kamu_flow_system_inmem::FlowEventStoreInMem>();
     b.add::<kamu_flow_system_inmem::FlowConfigurationEventStoreInMem>();

src/app/cli/src/explore/api_server.rs

@@ -12,7 +12,7 @@ use std::sync::Arc;
 
 use dill::Catalog;
 use internal_error::*;
-use kamu_flow_system_inmem::domain::{FlowService, FlowServiceRunConfig};
+use kamu_flow_system_inmem::domain::FlowService;
 use kamu_task_system_inmem::domain::TaskExecutor;
 
 /////////////////////////////////////////////////////////////////////////////////////////
@@ -97,12 +97,10 @@ impl APIServer {
     }
 
     pub async fn run(self) -> Result<(), InternalError> {
-        let flow_service_run_config = FlowServiceRunConfig::new(std::time::Duration::from_secs(1));
-
         tokio::select! {
             res = self.server => { res.int_err() },
             res = self.task_executor.run() => { res.int_err() },
-            res = self.flow_service.run(flow_service_run_config) => { res.int_err() }
+            res = self.flow_service.run() => { res.int_err() }
         }
     }
 }

src/domain/flow-system/src/services/flow/flow_service.rs

@@ -20,7 +20,7 @@ use crate::{DatasetFlowType, FlowID, FlowKey, FlowState, SystemFlowType};
 #[async_trait::async_trait]
 pub trait FlowService: Sync + Send {
     /// Runs the update main loop
-    async fn run(&self, run_config: FlowServiceRunConfig) -> Result<(), InternalError>;
+    async fn run(&self) -> Result<(), InternalError>;
 
     /// Triggers the specified flow manually, unless it's already waiting
     async fn trigger_manual_flow(
@@ -175,11 +175,11 @@ impl From<LoadError<FlowState>> for CancelFlowError {
 
 #[derive(Debug)]
 pub struct FlowServiceRunConfig {
-    pub awaiting_step: std::time::Duration,
+    pub awaiting_step: chrono::Duration,
 }
 
 impl FlowServiceRunConfig {
-    pub fn new(awaiting_step: std::time::Duration) -> Self {
+    pub fn new(awaiting_step: chrono::Duration) -> Self {
         Self { awaiting_step }
     }
 }

src/infra/flow-system-inmem/src/services/flow/flow_service_inmem.rs

@@ -11,7 +11,7 @@
 
 use std::sync::{Arc, Mutex};
 
-use chrono::{DateTime, Utc};
+use chrono::{DateTime, DurationRound, Utc};
 use dill::*;
 use event_bus::{AsyncEventHandler, EventBus};
 use futures::TryStreamExt;
@@ -30,6 +30,7 @@ use super::pending_flows_state::PendingFlowsState;
 
 pub struct FlowServiceInMemory {
     state: Arc<Mutex<State>>,
+    run_config: Arc<FlowServiceRunConfig>,
     event_bus: Arc<EventBus>,
     flow_event_store: Arc<dyn FlowEventStore>,
     time_source: Arc<dyn SystemTimeSource>,
@@ -57,6 +58,7 @@ struct State {
 #[scope(Singleton)]
 impl FlowServiceInMemory {
     pub fn new(
+        run_config: Arc<FlowServiceRunConfig>,
         event_bus: Arc<EventBus>,
         flow_event_store: Arc<dyn FlowEventStore>,
         time_source: Arc<dyn SystemTimeSource>,
@@ -66,6 +68,7 @@ impl FlowServiceInMemory {
     ) -> Self {
         Self {
             state: Arc::new(Mutex::new(State::default())),
+            run_config,
             event_bus,
             flow_event_store,
             time_source,
@@ -75,6 +78,13 @@ impl FlowServiceInMemory {
         }
     }
 
+    fn round_time(&self, time: DateTime<Utc>) -> Result<DateTime<Utc>, InternalError> {
+        let rounded_time = time
+            .duration_round(self.run_config.awaiting_step)
+            .int_err()?;
+        Ok(rounded_time)
+    }
+
     #[tracing::instrument(level = "debug", skip_all)]
     async fn run_current_timeslot(&self) {
         let planned_flows: Vec<_> = {
@@ -105,7 +115,10 @@ impl FlowServiceInMemory {
     }
 
     #[tracing::instrument(level = "debug", skip_all)]
-    async fn initialize_auto_polling_flows_from_configurations(&self) -> Result<(), InternalError> {
+    async fn initialize_auto_polling_flows_from_configurations(
+        &self,
+        start_time: DateTime<Utc>,
+    ) -> Result<(), InternalError> {
         let enabled_configurations: Vec<_> = self
             .flow_configuration_service
             .list_enabled_configurations()
@@ -114,8 +127,12 @@ impl FlowServiceInMemory {
             .int_err()?;
 
         for enabled_config in enabled_configurations {
-            self.activate_flow_configuration(enabled_config.flow_key, enabled_config.rule)
-                .await?;
+            self.activate_flow_configuration(
+                start_time,
+                enabled_config.flow_key,
+                enabled_config.rule,
+            )
+            .await?;
         }
 
         Ok(())
@@ -124,13 +141,15 @@ impl FlowServiceInMemory {
     #[tracing::instrument(level = "trace", skip_all, fields(?flow_key, ?rule))]
     async fn activate_flow_configuration(
         &self,
+        start_time: DateTime<Utc>,
         flow_key: FlowKey,
         rule: FlowConfigurationRule,
     ) -> Result<(), InternalError> {
         match &flow_key {
             FlowKey::Dataset(dataset_flow_key) => {
                 if let FlowConfigurationRule::Schedule(schedule) = &rule {
-                    self.enqueue_auto_polling_flow(&flow_key, schedule).await?;
+                    self.enqueue_auto_polling_flow(start_time, &flow_key, schedule)
+                        .await?;
                 }
 
                 let mut state = self.state.lock().unwrap();
@@ -140,7 +159,8 @@ impl FlowServiceInMemory {
             }
             FlowKey::System(system_flow_key) => {
                 if let FlowConfigurationRule::Schedule(schedule) = &rule {
-                    self.enqueue_auto_polling_flow(&flow_key, schedule).await?;
+                    self.enqueue_auto_polling_flow(start_time, &flow_key, schedule)
+                        .await?;
 
                     let mut state = self.state.lock().unwrap();
                     state
@@ -158,6 +178,7 @@ impl FlowServiceInMemory {
     #[tracing::instrument(level = "trace", skip_all, fields(?flow_key))]
     async fn try_enqueue_auto_polling_flow_if_enabled(
         &self,
+        start_time: DateTime<Utc>,
         flow_key: &FlowKey,
     ) -> Result<(), InternalError> {
         let maybe_active_schedule = self
@@ -168,7 +189,7 @@ impl FlowServiceInMemory {
             .try_get_flow_schedule(flow_key);
 
         if let Some(active_schedule) = maybe_active_schedule {
-            self.enqueue_auto_polling_flow(flow_key, &active_schedule)
+            self.enqueue_auto_polling_flow(start_time, flow_key, &active_schedule)
                 .await?;
         }
 
@@ -178,6 +199,7 @@ impl FlowServiceInMemory {
     #[tracing::instrument(level = "trace", skip_all, fields(?flow_key, ?schedule))]
     async fn enqueue_auto_polling_flow(
         &self,
+        start_time: DateTime<Utc>,
         flow_key: &FlowKey,
         schedule: &Schedule,
     ) -> Result<FlowState, InternalError> {
@@ -191,7 +213,7 @@ impl FlowServiceInMemory {
             None => {
                 let mut flow = self.make_new_flow(flow_key.clone(), trigger).await?;
 
-                let next_activation_time = schedule.next_activation_time(self.time_source.now());
+                let next_activation_time = schedule.next_activation_time(start_time);
                 self.enqueue_flow(flow.flow_id, next_activation_time)?;
 
                 flow.activate_at_time(self.time_source.now(), next_activation_time)
@@ -207,6 +229,7 @@ impl FlowServiceInMemory {
     #[tracing::instrument(level = "trace", skip_all, fields(%dataset_id, ?flow_type, %flow_id))]
     async fn enqueue_dependent_dataset_flows(
         &self,
+        start_time: DateTime<Utc>,
         dataset_id: &DatasetID,
         flow_type: DatasetFlowType,
         flow_id: FlowID,
@@ -259,15 +282,14 @@ impl FlowServiceInMemory {
                         if let Some(throttling_period) = start_condition.throttling_period {
                             // TODO: throttle not from NOW,
                             //  but from last flow of the dependent daataset
-                            let now = self.time_source.now();
                             self.enqueue_flow(
                                 dependent_dataset_flow.flow_id,
-                                now + throttling_period,
+                                start_time + throttling_period,
                             )?;
 
                             dependent_dataset_flow
                                 .define_start_condition(
-                                    now,
+                                    self.time_source.now(),
                                     FlowStartCondition::Throttling(FlowStartConditionThrottling {
                                         interval: throttling_period,
                                     }),
@@ -390,22 +412,24 @@ impl FlowServiceInMemory {
 impl FlowService for FlowServiceInMemory {
     /// Runs the update main loop
     #[tracing::instrument(level = "info", skip_all)]
-    async fn run(&self, run_config: FlowServiceRunConfig) -> Result<(), InternalError> {
+    async fn run(&self) -> Result<(), InternalError> {
         // Initial scheduling
-        self.initialize_auto_polling_flows_from_configurations()
+        let start_time = self.round_time(self.time_source.now())?;
+        self.initialize_auto_polling_flows_from_configurations(start_time)
             .await?;
 
         // Publish progress event
         self.event_bus
             .dispatch_event(FlowServiceEventConfigurationLoaded {
-                event_time: self.time_source.now(),
+                event_time: start_time,
             })
             .await
             .int_err()?;
 
         // Main scanning loop
         let main_loop_span = tracing::debug_span!("FlowService main loop");
         let _ = main_loop_span.enter();
+        let std_awaiting_step = self.run_config.awaiting_step.to_std().int_err()?;
 
         loop {
             // Do we have a timeslot scheduled?
@@ -425,13 +449,13 @@ impl FlowService for FlowServiceInMemory {
                 // Publish progress event
                 self.event_bus
                     .dispatch_event(FlowServiceEventExecutedTimeSlot {
-                        event_time: current_time,
+                        event_time: nearest_activation_time,
                     })
                     .await
                     .int_err()?;
             }
 
-            tokio::time::sleep(run_config.awaiting_step).await;
+            tokio::time::sleep(std_awaiting_step).await;
             continue;
         }
     }
@@ -664,6 +688,7 @@ impl AsyncEventHandler<TaskEventFinished> for FlowServiceInMemory {
                     && flow_key.flow_type.is_dataset_update()
                 {
                     self.enqueue_dependent_dataset_flows(
+                        self.round_time(event.event_time)?,
                         &flow_key.dataset_id,
                         flow_key.flow_type,
                         flow.flow_id,
@@ -681,7 +706,7 @@ impl AsyncEventHandler<TaskEventFinished> for FlowServiceInMemory {
             // In case of success:
             //  - enqueue next auto-polling flow cycle
             if event.outcome == TaskOutcome::Success {
-                self.try_enqueue_auto_polling_flow_if_enabled(&flow.flow_key)
+                self.try_enqueue_auto_polling_flow_if_enabled(event.event_time, &flow.flow_key)
                     .await?;
             }
 
@@ -703,8 +728,12 @@ impl AsyncEventHandler<FlowConfigurationEventModified> for FlowServiceInMemory {
             state.active_configs.drop_flow_config(&event.flow_key);
             // TODO: should we unqueue pending flows / abort scheduled tasks?
         } else {
-            self.activate_flow_configuration(event.flow_key.clone(), event.rule.clone())
-                .await?
+            self.activate_flow_configuration(
+                self.round_time(event.event_time)?,
+                event.flow_key.clone(),
+                event.rule.clone(),
+            )
+            .await?
         }
 
         Ok(())

src/infra/flow-system-inmem/tests/tests/test_flow_service_inmem.rs

@@ -47,11 +47,9 @@ async fn test_read_initial_config() {
         )
         .await;
 
-    let config = FlowServiceRunConfig::new(std::time::Duration::from_millis(5));
-
     let _ = tokio::select! {
-        res = harness.flow_service.run(config) => res.int_err(),
-        _ = tokio::time::sleep(std::time::Duration::from_millis(50)) => Ok(()),
+        res = harness.flow_service.run() => res.int_err(),
+        _ = tokio::time::sleep(std::time::Duration::from_millis(60)) => Ok(()),
     }
     .unwrap();
 
@@ -64,13 +62,12 @@ async fn test_read_initial_config() {
     assert_eq!(3, state.snapshots.len());
 
     let start_moment = state.snapshots[0].0;
-    let foo_scheduled_moment = state.snapshots[1].0;
-    let bar_scheduled_moment = state.snapshots[2].0;
+    let foo_moment = state.snapshots[1].0;
+    let bar_moment = state.snapshots[2].0;
 
-    assert!(start_moment < foo_scheduled_moment && foo_scheduled_moment < bar_scheduled_moment);
-    assert!((foo_scheduled_moment - start_moment) >= Duration::milliseconds(30));
-    assert!((foo_scheduled_moment - start_moment) < Duration::milliseconds(45));
-    assert!((bar_scheduled_moment - start_moment) >= Duration::milliseconds(45));
+    assert!(start_moment < foo_moment && foo_moment < bar_moment);
+    assert_eq!((foo_moment - start_moment), Duration::milliseconds(30));
+    assert_eq!((bar_moment - start_moment), Duration::milliseconds(45));
 
     let flow_test_checks = [
         // Snapshot 0: after initial queueing
@@ -258,6 +255,7 @@ impl FlowHarness {
 
         let catalog = dill::CatalogBuilder::new()
             .add::<EventBus>()
+            .add_value(FlowServiceRunConfig::new(Duration::milliseconds(5)))
             .add::<FlowServiceInMemory>()
             .add::<FlowEventStoreInMem>()
             .add::<FlowConfigurationServiceInMemory>()