Greptile Summary

This PR addresses two critical issues with derived stream scheduling when delay configurations are involved:

Issue 1: Invalid Time Ranges
When a derived stream has a non-zero delay configured, the system sometimes calculates a start time that is greater than the end time, causing processing failures. The fix introduces time alignment logic and skip handling in the scheduler. The supposed_to_be_run_at timestamp is now aligned to ensure consistent time boundaries, and when an invalid time range is detected (start > end), the scheduler logs a warning and moves to the next valid run window instead of attempting to process the invalid range.

Issue 2: Duplicate Results from Updates
Previously, updating a scheduled derived stream would always delete the existing trigger and recreate it, causing the start time context to be lost and potentially leading to duplicate results. The solution implements conditional trigger deletion - triggers are only deleted when the pipeline source type actually changes from scheduled to realtime. For scheduled-to-scheduled updates, the existing trigger is preserved and only its next_run_at field is updated.

Implementation Details
The changes span three files: adding helper methods (is_scheduled() and is_realtime()) to the PipelineSource enum for cleaner type checking, implementing comprehensive skip logic in the scheduler handlers with improved logging and early TriggerData initialization, and modifying the pipeline update logic to conditionally preserve existing triggers. The scheduler now properly handles catch-up scenarios using aligned timestamps and reports skipped runs with appropriate status tracking.

Confidence score: 4/5

• This PR addresses well-defined scheduling issues with a targeted approach that preserves existing functionality
• Score reflects complex timing logic in critical scheduler code that handles edge cases around delays and time boundaries
• Pay close attention to the scheduler handlers file where most of the complex time calculation and skip logic is implemented

_{3 files reviewed, no comments}

_{Edit Code Review Bot Settings | Greptile}