Scheduling & Triggers

Flyte manages automated task execution through a robust scheduling system that combines persistent state management with an in-memory cron engine. This architecture ensures that schedules are not only executed on time but also remain consistent across worker restarts and handle missed runs gracefully.

In this tutorial, you will learn how to manage triggers using the TriggerRepo, initialize the scheduler worker, and understand how Flyte ensures idempotent execution of scheduled runs.

Prerequisites

To follow this tutorial, you should be familiar with the Flyte data model, specifically models.Trigger. You will also need access to a database context and the Flyte repository layer.

Step 1: Persisting a Trigger

The foundation of scheduling in Flyte is the TriggerRepo. It manages the triggers table (latest state) and the trigger_revisions table (history). When you save a trigger, Flyte also denormalizes metadata back to the tasks table to keep task-level statistics in sync.

Use the SaveTrigger method to create or update a schedule. Note the use of expectedRevision for optimistic locking.

import (
    "context"
    "github.com/flyteorg/flyte/v2/runs/repository/interfaces"
    "github.com/flyteorg/flyte/v2/runs/repository/models"
)

func createCronTrigger(ctx context.Context, repo interfaces.TriggerRepo) error {
    trigger := &models.Trigger{
        Project:    "flytesnacks",
        Domain:     "development",
        TaskName:   "my_task",
        Name:       "daily_sync",
        CronString: "0 0 * * *", // Run every day at midnight
        Active:     true,
    }

    // Pass 0 as expectedRevision for a brand-new trigger
    _, err := repo.SaveTrigger(ctx, trigger, 0)
    if err != nil {
        return err
    }
    return nil
}

The SaveTrigger implementation in runs/repository/impl/trigger.go wraps the upsert and the metadata refresh in a single transaction to ensure consistency.

Step 2: Initializing the Scheduler Worker

The scheduler worker is the component that monitors the database for active triggers and schedules them for execution. You initialize it using the Start function found in runs/scheduler/start.go.

This function wires together the TriggerExecutor, the GoCronScheduler, and the ScheduleSyncer.

import (
    "context"
    "github.com/flyteorg/flyte/v2/runs/scheduler"
    "github.com/flyteorg/flyte/v2/runs/scheduler/config"
)

func startScheduler(ctx context.Context, repo interfaces.TriggerRepo) {
    cfg := config.TriggerSchedulerConfig{
        Enabled:               true,
        ResyncInterval:        30 * time.Second,
        MaxCatchupRunsPerLoop: 100,
        ExecutionQPS:          10.0,
        ExecutionBurst:        20,
    }

    // Start returns a function that runs the scheduler loop
    runFunc := scheduler.Start(ctx, repo, cfg, "http://localhost:8080")
    
    go func() {
        if err := runFunc(ctx); err != nil {
            // Handle fatal scheduler error
        }
    }()
}

Step 3: Synchronization and Catchup

When the scheduler starts, it performs a "bootstrap" phase. It loads all active triggers from the database and checks if any runs were missed while the scheduler was offline. This is handled by CatchupAll.

The ScheduleSyncer then takes over, periodically polling the TriggerRepo to reconcile the in-memory robfig/cron jobs with the database state.

// Inside runs/scheduler/start.go
triggers, err := core.ListActiveScheduleTriggers(ctx, triggerRepo)
if err != nil {
    return err
}

// Update the in-memory cron jobs
sched.UpdateSchedules(ctx, triggers)

// Fire missed runs since the last recorded TriggeredAt or UpdatedAt
sched.CatchupAll(ctx, triggers, time.Now().UTC(), cfg.MaxCatchupRunsPerLoop)

// Start the periodic resync loop
syncer := core.NewScheduleSyncer(triggerRepo, sched, cfg.ResyncInterval)
err = syncer.Run(ctx)

The scheduler uses the latest of TriggeredAt or UpdatedAt as the baseline for scheduling to avoid double-firing runs that occurred just before a trigger was deactivated or updated.

Step 4: Execution and Idempotency

When a cron job fires, the TriggerExecutor (defined in runs/scheduler/executor/trigger_executor.go) is responsible for calling the RunService to create a new execution.

To prevent duplicate runs in a distributed environment or during retries, Flyte generates a deterministic run name based on the trigger identity and the scheduled time.

// 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())
}

If the CreateRun call returns an AlreadyExists error, the TriggerExecutor treats it as a success, ensuring that each scheduled slot is executed exactly once.

Summary

By using the TriggerRepo for persistence and the GoCronScheduler for execution, Flyte provides a reliable scheduling system. The deterministic naming in TriggerExecutor ensures idempotency, while the ScheduleSyncer keeps the system responsive to configuration changes in the database.

For next steps, explore the TriggerSchedulerConfig to tune the ExecutionQPS and ResyncInterval for your deployment's scale.