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"] ``` <details> <summary><h3> File Walkthrough</h3></summary> <table><thead><tr><th></th><th align="left">Relevant files</th></tr></thead><tbody><tr><td><strong>Enhancement</strong></td><td><table> <tr> <td> <details> <summary><strong>components.rs</strong><dd><code>PipelineSource helpers for source type checks</code>                        </dd></summary> <hr> src/config/src/meta/pipeline/components.rs <ul><li>Add helper methods <code>is_scheduled</code> and <code>is_realtime</code><br> <li> Keep existing default implementation</ul> </details> </td> <td><a href="https://github.com/openobserve/openobserve/pull/8064/files#diff-e2d647c422ae05ac91abc683c7b11a3fb7da61ee30968ca6e84cdba40221bbb9">+10/-0</a>    </td> </tr> <tr> <td> <details> <summary><strong>mod.rs</strong><dd><code>Safer trigger deletion on pipeline source change</code>                  </dd></summary> <hr> src/service/pipeline/mod.rs <ul><li>Delete previous scheduled trigger only when source changes<br> <li> Add use of <code>is_realtime</code> guard during updates</ul> </details> </td> <td><a href="https://github.com/openobserve/openobserve/pull/8064/files#diff-c186c796a7bdfd9d1301b1429df11456430f1d0d4758783810b9e2e44384e3b2">+12/-9</a>    </td> </tr> </table></td></tr><tr><td><strong>Bug fix</strong></td><td><table> <tr> <td> <details> <summary><strong>handlers.rs</strong><dd><code>Robust delayed scheduling and alignment fixes</code>                        </dd></summary> <hr> src/service/alerts/scheduler/handlers.rs <ul><li>Defer runs by 7 days for expired free-trial orgs<br> <li> Compute aligned end and supposed run times<br> <li> Skip runs with invalid start/end due to delay; set Skipped<br> <li> Reorder retry handling; adjust next_run and logging<br> <li> Update trigger_data start/end based on alignment<br> <li> Use aligned time for catch-up logic</ul> </details> </td> <td><a href="https://github.com/openobserve/openobserve/pull/8064/files#diff-5eca8995a88fcecb6d6b8efd3c7c4496f4a96a4b060fb4ed85593574f7cf49eb">+89/-39</a>  </td> </tr> </table></td></tr></tr></tbody></table> </details> ___
2026-11-05 08:09:22 +00:00 · 2025-09-12 14:18:22 +05:30
parent e7b7fe5f03
commit 6c848afcab
3 changed files with 150 additions and 88 deletions
@@ -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)]
@@ -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 {
@@ -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
            }