Core Scheduling Architecture

Flyte uses a robust scheduling system to manage time-based triggers for workflows. This system ensures that scheduled tasks are executed reliably, even across service restarts or database updates, by reconciling the desired state in the database with an in-memory cron scheduler.

Scheduling Architecture Overview

The scheduling system in Flyte is composed of three primary layers that move data from the database to execution:

1. ScheduleSyncer: A background worker that periodically polls the database for active triggers.
2. GoCronScheduler: A manager that maintains a set of running cron jobs, reconciling them whenever the syncer finds changes.
3. GoCronJob: A wrapper around the robfig/cron library that executes a specific trigger at its scheduled time.

This architecture follows a "reconciliation" pattern rather than a simple event-driven one. Instead of reacting to every database change, the system periodically aligns its internal state with the truth in the database.

Managing the Scheduler Lifecycle

The GoCronScheduler (found in runs/scheduler/core/scheduler.go) is the central engine. It wraps the robfig/cron library and manages a map of active jobs.

Reconciliation with UpdateSchedules

When the ScheduleSyncer fetches the latest active triggers, it calls UpdateSchedules. This method performs a three-way reconciliation:

• Removal: Jobs for triggers that are no longer active or have been deleted are stopped and removed from the scheduler.
• Updates: If a trigger's LatestRevision has changed (indicating a change in the cron expression or fixed rate), the existing job is replaced with a new one.
• Addition: New active triggers are registered as new cron jobs.

// From runs/scheduler/core/scheduler.go
func (s *GoCronScheduler) UpdateSchedules(ctx context.Context, triggers []*models.Trigger) {
    // ... (logic to determine desired state) ...

    s.mu.Lock()
    defer s.mu.Unlock()

    // Remove jobs no longer desired.
    for key, job := range s.jobs {
        if _, ok := desired[key]; !ok {
            s.cron.Remove(job.entryID)
            delete(s.jobs, key)
        }
    }

    // Add new jobs or update changed ones.
    for key, t := range desired {
        if existing, exists := s.jobs[key]; exists {
            if existing.trigger.LatestRevision == t.LatestRevision {
                continue
            }
            s.cron.Remove(existing.entryID)
        }
        // ... (logic to create and schedule NewGoCronJob) ...
    }
}

Handling Downtime with CatchupAll

If the scheduler service is down, it may miss several scheduled execution windows. To prevent data gaps, Flyte includes a "catchup" mechanism. During bootstrap, the scheduler calculates missed runs by comparing the current time with the trigger's last execution time (TriggeredAt) or activation time (UpdatedAt).

The CatchupAll method iterates through active triggers and fires executions for these missed slots, subject to a configurable limit (MaxCatchupRunsPerLoop) to prevent overwhelming the system.

The Execution Loop

Each scheduled job is represented by a GoCronJob. When the cron expression matches the current time, the robfig/cron library calls the Run method on the job.

Deterministic Execution and Idempotency

A critical requirement for the scheduler is idempotency. If a job fires twice for the same scheduled time (e.g., during a catchup loop and a normal cron fire), Flyte must not create duplicate workflow runs.

The TriggerExecutor (in runs/scheduler/executor/trigger_executor.go) solves this by generating a deterministic run name. This name is a hash of the project, domain, task name, trigger name, and the exact scheduled timestamp.

// From runs/scheduler/executor/trigger_executor.go
func runName(t *models.Trigger, scheduledAt time.Time) string {
    h := fnv.New64()
    _, _ = fmt.Fprintf(h, "%s:%s:%s:%s:%d:%d:%d:%d:%d:%d",
        t.Project, t.Domain, t.TaskName, t.Name,
        scheduledAt.Year(), scheduledAt.Month(), scheduledAt.Day(),
        scheduledAt.Hour(), scheduledAt.Minute(), scheduledAt.Second())
    return fmt.Sprintf("r%x", h.Sum64())
}

When the executor calls the Run Service to create the run, it uses this deterministic name. If the run already exists, the service returns an AlreadyExists error, which the TriggerExecutor treats as a successful execution.

Periodic Synchronization

The ScheduleSyncer (in runs/scheduler/core/schedule_syncer.go) runs a continuous loop that triggers the reconciliation process. It uses a ticker (defaulting to 30 seconds) to fetch all active triggers from the database.

// From runs/scheduler/core/schedule_syncer.go
func (s *ScheduleSyncer) Run(ctx context.Context) error {
    s.sync(ctx) // Immediate sync on startup

    ticker := time.NewTicker(s.resyncInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-ticker.C:
            s.sync(ctx)
        }
    }
}

Configuration

You can tune the behavior of the scheduling system via the TriggerSchedulerConfig. These settings control how often the system syncs and how aggressively it catches up on missed runs.

Parameter	Default	Description
resyncInterval	30s	How often the ScheduleSyncer polls the database for active triggers.
maxCatchupRunsPerLoop	100	The maximum number of missed runs the scheduler will fire during a single catchup cycle.
executionQps	10.0	The rate limit (queries per second) for the TriggerExecutor when calling the Run Service.
executionBurst	20	The burst capacity for the executor's rate limiter.

These configurations are typically managed in the Flyte deployment configuration under the runs.triggerScheduler section.