From 6c848afcab9e0443a8d28eaeddee6f358123d3bf Mon Sep 17 00:00:00 2001 From: Subhra264 Date: Fri, 12 Sep 2025 14:18:22 +0530 Subject: [PATCH] fix: skip when start time > end time for delayed derived stream (#8064) ### **User description** - When delay is non-zero for derived stream, currently sometimes the start time becomes greater than the end time. This pr fixes that by moving to the next run where the end time eventually becomes greater than the start time and also follows the given delay. - Currently when updating a derived stream, it deletes the old scheduled job, and hence the start time for the scheduled job changes. As a result, there can be duplicate results in the derived stream because of updates. This pr fixes this by just updating only the next_run_at field of the scheduled_job. ___ ### **PR Type** Bug fix, Enhancement ___ ### **Description** - Prevent invalid time ranges with delay - Align end/scheduled times consistently - Respect free-trial by deferring runs a week - Avoid deleting triggers on updates ___ ### Diagram Walkthrough ```mermaid flowchart LR A["DerivedStream trigger handling"] -- "align end time" --> B["aligned_end_time computed"] B -- "start <= end ?" --> C["Proceed with eval"] B -- "start > end" --> D["Skip to next run"] D -- "set Skipped status" --> E["update trigger + usage"] C -- "evaluate + set times" --> F["update trigger/next_run"] G["Free trial check"] -- "expired" --> H["defer next_run_at by 7 days"] I["Pipeline update"] -- "Scheduled -> Realtime" --> J["delete old trigger only on source change"] ```

File Walkthrough
Relevant files
Enhancement
components.rs
PipelineSource helpers for source type checks                       
src/config/src/meta/pipeline/components.rs
  • Add helper methods is_scheduled and is_realtime
  • Keep existing default implementation
 +10/-0   
mod.rs
Safer trigger deletion on pipeline source change                 
src/service/pipeline/mod.rs
  • Delete previous scheduled trigger only when source changes
  • Add use of is_realtime guard during updates
 +12/-9   
Bug fix
handlers.rs
Robust delayed scheduling and alignment fixes                       
src/service/alerts/scheduler/handlers.rs
  • Defer runs by 7 days for expired free-trial orgs
  • Compute aligned end and supposed run times
  • Skip runs with invalid start/end due to delay; set Skipped
  • Reorder retry handling; adjust next_run and logging
  • Update trigger_data start/end based on alignment
  • Use aligned time for catch-up logic
 +89/-39 
___ --- src/config/src/meta/pipeline/components.rs | 10 + src/service/alerts/scheduler/handlers.rs | 207 +++++++++++++-------- src/service/pipeline/mod.rs | 21 ++- 3 files changed, 150 insertions(+), 88 deletions(-) diff --git a/src/config/src/meta/pipeline/components.rs b/src/config/src/meta/pipeline/components.rs index 2febc4b6bc..6309113ef3 100644 --- a/src/config/src/meta/pipeline/components.rs +++ b/src/config/src/meta/pipeline/components.rs @@ -38,6 +38,16 @@ impl Default for PipelineSource { } } +impl PipelineSource { + pub fn is_scheduled(&self) -> bool { + matches!(self, Self::Scheduled(_)) + } + + pub fn is_realtime(&self) -> bool { + matches!(self, Self::Realtime(_)) + } +} + #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Default, ToSchema)] #[serde(rename_all = "snake_case")] #[serde(default)] diff --git a/src/service/alerts/scheduler/handlers.rs b/src/service/alerts/scheduler/handlers.rs index 7815f272bc..7f85fbafea 100644 --- a/src/service/alerts/scheduler/handlers.rs +++ b/src/service/alerts/scheduler/handlers.rs @@ -1085,15 +1085,57 @@ async fn handle_derived_stream_triggers( { if !is_org_in_free_trial_period(&trigger.org).await? { log::info!( - "skipping pipeline {} id {} in org {} because free trial expiry", + "[Scheduler trace_id {scheduler_trace_id}] Skipping pipeline {} id {} in org {} because free trial expiry", pipeline.name, pipeline_id, trigger.org ); + new_trigger.next_run_at += Duration::try_days(7).unwrap().num_microseconds().unwrap(); + db::scheduler::update_trigger(new_trigger).await?; return Ok(()); } } + if !pipeline.enabled { + // Pipeline not enabled, check again in 5 mins + let msg = format!( + "Pipeline associated with trigger not enabled: {org_id}/{stream_type}/{pipeline_name}/{pipeline_id}. Checking after 5 mins." + ); + new_trigger.next_run_at += Duration::try_minutes(5) + .unwrap() + .num_microseconds() + .unwrap(); + let trigger_data_stream = TriggerData { + _timestamp: now_micros(), + org: new_trigger.org.clone(), + module: TriggerDataType::DerivedStream, + key: new_trigger.module_key.clone(), + next_run_at: new_trigger.next_run_at, + is_realtime: new_trigger.is_realtime, + is_silenced: new_trigger.is_silenced, + status: TriggerDataStatus::Failed, + start_time: 0, + end_time: 0, + retries: new_trigger.retries, + error: Some(msg.clone()), + success_response: None, + is_partial: None, + delay_in_secs: None, + evaluation_took_in_secs: None, + source_node: Some(LOCAL_NODE.name.clone()), + query_took: None, + scheduler_trace_id: Some(scheduler_trace_id.clone()), + time_in_queue_ms: Some(time_in_queue), + skipped_alerts_count: None, + }; + log::info!("[SCHEDULER trace_id {scheduler_trace_id}] {msg}"); + new_trigger_data.reset(); + new_trigger.data = new_trigger_data.to_json_string(); + db::scheduler::update_trigger(new_trigger).await?; + publish_triggers_usage(trigger_data_stream).await; + return Ok(()); + } + let Some(derived_stream) = pipeline.get_derived_stream() else { let err_msg = format!( "DerivedStream associated with the trigger not found in pipeline: {org_id}/{pipeline_name}/{pipeline_id}. Checking after 5 mins." @@ -1156,6 +1198,17 @@ async fn handle_derived_stream_triggers( .unwrap() .num_microseconds() .unwrap(); + let aligned_supposed_to_be_run_at = if !is_cron_frequency { + TriggerCondition::align_time( + supposed_to_be_run_at, + derived_stream.tz_offset, + Some(derived_stream.trigger_condition.period * 60), + ) + } else { + // For cron frequency, we don't need to align the end time as it is already aligned (the + // cron crate takes care of it) + TriggerCondition::align_time(supposed_to_be_run_at, derived_stream.tz_offset, None) + }; let (mut start, mut end) = if derived_stream.start_at.is_some() && trigger.data.is_empty() { (derived_stream.start_at, supposed_to_be_run_at) @@ -1170,7 +1223,8 @@ async fn handle_derived_stream_triggers( // data for the period from 5:05pm to 5:15pm. Which is to cover the skipped // period from 5:05pm to 5:15pm. log::debug!( - "supposed_to_be_run_at: {}, t0 + supposed_to_be_run_at: {}, supposed_to_be_run_smaller: {}", + "[SCHEDULER trace_id {scheduler_trace_id}] module key: {}, supposed_to_be_run_at: {}, t0 + supposed_to_be_run_at: {}, supposed_to_be_run_smaller: {}", + new_trigger.module_key, chrono::DateTime::from_timestamp_micros(supposed_to_be_run_at) .unwrap() .time(), @@ -1202,7 +1256,7 @@ async fn handle_derived_stream_triggers( (None, supposed_to_be_run_at) }; // For derived stream, period is in minutes, so we need to convert it to seconds for align_time - let aligned_curr_time = if !is_cron_frequency { + let aligned_end_time = if !is_cron_frequency { // For non-cron frequency, we need to align the current time so that the end_time is // divisible by the period For example, if the current time is 5:19pm, period is 5 // mins, and delay is 4mins (supposed to be run at 5:15pm), we need to ingest data @@ -1220,81 +1274,6 @@ async fn handle_derived_stream_triggers( // cron crate takes care of it) TriggerCondition::align_time(end, derived_stream.tz_offset, None) }; - // conditionally modify supposed_to_be_run_at - if start.is_none_or(|t0| t0 < aligned_curr_time) { - end = aligned_curr_time; - } - - if !pipeline.enabled { - // Pipeline not enabled, check again in 5 mins - let msg = format!( - "Pipeline associated with trigger not enabled: {org_id}/{stream_type}/{pipeline_name}/{pipeline_id}. Checking after 5 mins." - ); - new_trigger.next_run_at += Duration::try_minutes(5) - .unwrap() - .num_microseconds() - .unwrap(); - let trigger_data_stream = TriggerData { - _timestamp: now_micros(), - org: new_trigger.org.clone(), - module: TriggerDataType::DerivedStream, - key: new_trigger.module_key.clone(), - next_run_at: new_trigger.next_run_at, - is_realtime: new_trigger.is_realtime, - is_silenced: new_trigger.is_silenced, - status: TriggerDataStatus::Skipped, - start_time: 0, - end_time: 0, - retries: new_trigger.retries, - error: Some(msg.clone()), - success_response: None, - is_partial: None, - delay_in_secs: None, - evaluation_took_in_secs: None, - source_node: Some(LOCAL_NODE.name.clone()), - query_took: None, - scheduler_trace_id: Some(scheduler_trace_id.clone()), - time_in_queue_ms: Some(time_in_queue), - skipped_alerts_count: None, - }; - log::info!("[SCHEDULER trace_id {scheduler_trace_id}] {msg}"); - // new_trigger_data.reset(); - // new_trigger.data = new_trigger_data.to_json_string(); - db::scheduler::update_trigger(new_trigger).await?; - publish_triggers_usage(trigger_data_stream).await; - return Ok(()); - } - - // In case the scheduler background job (watch_timeout) updates the trigger retries - // (not through this handler), we need to skip to the next run at but with the same - // trigger start time. If we don't handle here, in that case, the `clean_complete` - // background job will clear this job as it has reached max retries. - if trigger.retries >= max_retries { - log::info!( - "[SCHEDULER trace_id {scheduler_trace_id}] DerivedStream trigger: {}/{} has reached maximum possible retries. Skipping to next run", - new_trigger.org, - new_trigger.module_key - ); - // Go to the next nun at, but use the same trigger start time - new_trigger.next_run_at = derived_stream - .trigger_condition - .get_next_trigger_time_non_aligned( - false, - derived_stream.tz_offset, - false, - Some(trigger.next_run_at), - )?; - // Start over next time - new_trigger.retries = 0; - db::scheduler::update_trigger(new_trigger).await?; - return Ok(()); - } - - log::debug!( - "[SCHEDULER trace_id {scheduler_trace_id}] DerivedStream: querying for time range: start_time {}, end_time {}.", - start.unwrap_or_default(), - end, - ); let mut trigger_data_stream = TriggerData { _timestamp: now_micros(), @@ -1324,6 +1303,76 @@ async fn handle_derived_stream_triggers( skipped_alerts_count: None, }; + // conditionally modify supposed_to_be_run_at + if start.is_none_or(|t0| t0 < aligned_end_time) { + end = aligned_end_time; + } else { + // either t0 = aligned_end_time or t0 > aligned_end_time + // in both cases, we need to skip to next run because, we should always use + // aligned_curr_time as the end time + // Invalid timerange, most probably due to non-zero delay + // Don't do any further processing, just skip to next run + let start_time = start.unwrap(); + log::warn!( + "[SCHEDULER trace_id {scheduler_trace_id}] module key: {}, Invalid timerange. Skipping to next run. start: {}, end: {}", + new_trigger.module_key, + start_time, + end, + ); + new_trigger.next_run_at = derived_stream.trigger_condition.get_next_trigger_time( + false, + derived_stream.tz_offset, + false, + Some(trigger.next_run_at), + )?; + trigger_data_stream.status = TriggerDataStatus::Skipped; + trigger_data_stream.end_time = aligned_end_time; + trigger_data_stream.next_run_at = new_trigger.next_run_at; + trigger_data_stream.start_time = start_time; + trigger_data_stream.error = Some(format!( + "Invalid timerange - start: {start_time}, end: {end}, should be fixed in the next run" + )); + db::scheduler::update_trigger(new_trigger).await?; + publish_triggers_usage(trigger_data_stream).await; + return Ok(()); + } + + // In case the scheduler background job (watch_timeout) updates the trigger retries + // (not through this handler), we need to skip to the next run at but with the same + // trigger start time. If we don't handle here, in that case, the `clean_complete` + // background job will clear this job as it has reached max retries. + if trigger.retries >= max_retries { + log::info!( + "[SCHEDULER trace_id {scheduler_trace_id}] DerivedStream trigger: {}/{} has reached maximum possible retries. Skipping to next run", + new_trigger.org, + new_trigger.module_key + ); + // Go to the next nun at, but use the same trigger start time + new_trigger.next_run_at = derived_stream.trigger_condition.get_next_trigger_time( + false, + derived_stream.tz_offset, + false, + Some(trigger.next_run_at), + )?; + // Start over next time + new_trigger.retries = 0; + db::scheduler::update_trigger(new_trigger).await?; + return Ok(()); + } + + log::debug!( + "[SCHEDULER trace_id {scheduler_trace_id}] DerivedStream: {} querying for time range: start_time {}, end_time {}.", + new_trigger.module_key, + start.unwrap_or_default(), + end, + ); + + // end can change due to delay feature, so we need to update the start and end time + if start.is_none() { + trigger_data_stream.start_time = end - period_num_microseconds; + } + trigger_data_stream.end_time = end; + // evaluate trigger and configure trigger next run time match derived_stream .evaluate( @@ -1551,7 +1600,7 @@ async fn handle_derived_stream_triggers( if !(trigger_data_stream.status == TriggerDataStatus::Failed && new_trigger.retries < max_retries) { - let need_to_catch_up = end < supposed_to_be_run_at; + let need_to_catch_up = end < aligned_supposed_to_be_run_at; // If the trigger didn't fail, we need to reset the `retries` count. // Only cumulative failures should be used to check with `max_retries` if trigger_data_stream.status != TriggerDataStatus::Failed { diff --git a/src/service/pipeline/mod.rs b/src/service/pipeline/mod.rs index de968e7d2c..4969b32e09 100644 --- a/src/service/pipeline/mod.rs +++ b/src/service/pipeline/mod.rs @@ -110,16 +110,19 @@ pub async fn update_pipeline(mut pipeline: Pipeline) -> Result<(), PipelineError match existing_pipeline.source { // realtime: remove prev. src. stream_params from cache PipelineSource::Realtime(stream_params) => Some(stream_params), - // scheduled: delete prev. trigger + // scheduled: delete prev. trigger if source has changed PipelineSource::Scheduled(derived_stream) => { - if let Err(error) = super::alerts::derived_streams::delete( - &derived_stream, - &existing_pipeline.name, - &existing_pipeline.id, - ) - .await - { - return Err(PipelineError::DeleteDerivedStream(error.to_string())); + if pipeline.source.is_realtime() { + // source changed, delete prev. trigger + if let Err(error) = super::alerts::derived_streams::delete( + &derived_stream, + &existing_pipeline.name, + &existing_pipeline.id, + ) + .await + { + return Err(PipelineError::DeleteDerivedStream(error.to_string())); + } } None }