Automating Workflows with Triggers

Flyte automates workflow executions through a trigger system that supports both event-based and scheduled (cron) activations. The system is designed for high auditability and consistency, maintaining a full revision history for every trigger and using optimistic locking to prevent concurrent update conflicts.

Managing Triggers

You manage triggers primarily through the triggerService (found in runs/service/trigger_service.go). When you deploy a trigger, the service validates the request and persists the state using the TriggerRepo.

Deploying a Trigger

To create or update a trigger, you send a DeployTriggerRequest. Internally, the service converts the Protobuf spec into a models.Trigger and calls SaveTrigger.

// From runs/service/trigger_service.go
func (s *triggerService) DeployTrigger(
	ctx context.Context,
	req *connect.Request[triggerpb.DeployTriggerRequest],
) (*connect.Response[triggerpb.DeployTriggerResponse], error) {
	request := req.Msg
    // ... validation ...
	triggerModel, err := transformers.NewTriggerModel(ctx, id, request.GetSpec(), request.GetAutomationSpec())
	saved, err := s.db.TriggerRepo().SaveTrigger(ctx, triggerModel, request.GetRevision())
    // ...
}

The SaveTrigger method in runs/repository/impl/trigger.go performs an "upsert" operation. If the trigger already exists (identified by the composite key of project, domain, task name, and name), it updates the existing row and increments the latest_revision.

Optimistic Locking

To prevent two users from overwriting each other's changes, Flyte uses an optimistic locking pattern. When updating an existing trigger, you must provide the expectedRevision (the version you last read).

If the latest_revision in the database has changed since you read it, the update will fail with an "optimistic lock failure" error.

// From runs/repository/impl/trigger.go
if expectedRevision > 0 {
    suffix += fmt.Sprintf(" WHERE triggers.latest_revision = %d", expectedRevision)
}
// ...
if errors.Is(err, sql.ErrNoRows) {
    return errors.New("optimistic lock failure: trigger was modified concurrently, please fetch latest and retry")
}

Revision History and Auditability

Flyte maintains a strict separation between the current state of a trigger and its history.

1. triggers table: Stores the latest state of each trigger. It is updated in-place.
2. trigger_revisions table: An append-only table that stores an immutable snapshot of the trigger every time it is modified.

Every action—whether it is a DEPLOY, ACTIVATE, DEACTIVATE, or DELETE—results in a new row in the trigger_revisions table. This allows you to inspect the history of a trigger and see exactly who changed it and when.

// From runs/repository/models/trigger.go
type TriggerRevision struct {
	Project  string `db:"project"`
	Domain   string `db:"domain"`
	TaskName string `db:"task_name"`
	Name     string `db:"name"`
	Revision uint64 `db:"revision"` // Part of the composite PK
    // ... snapshot fields ...
	Action   string `db:"action"`   // TRIGGER_REVISION_ACTION_DEPLOY, etc.
}

Task Integration and Visibility

To ensure that the Task service can quickly display whether a task has active triggers without performing complex joins, Flyte denormalizes trigger metadata onto the tasks table.

Whenever a trigger is saved, updated, or deleted, the TriggerRepo invokes refreshTaskTriggerMeta within the same database transaction. This function recomputes the summary statistics for the associated task:

• total_triggers: Total number of triggers attached to the task version.
• active_triggers: Number of triggers currently in an active state.
• trigger_name: The name of the trigger (if exactly one is attached).

// From runs/repository/impl/trigger.go
func refreshTaskTriggerMeta(ctx context.Context, tx *sqlx.Tx, t *models.Trigger) error {
    // ... computes stats from triggers table ...
	result, err := tx.ExecContext(ctx, `
		UPDATE tasks SET
			trigger_name            = $1,
			total_triggers          = $2,
			active_triggers         = $3,
			trigger_automation_spec = $4
		WHERE project = $5 AND domain = $6 AND name = $7 AND version = $8`,
		triggerName, totalTriggers, activeTriggers, automationSpec,
		t.Project, t.Domain, t.TaskName, t.TaskVersion,
	)
    // ...
}

The Scheduler Worker

The GoCronScheduler is the background worker responsible for executing scheduled triggers. It does not rely on a persistent connection to the database; instead, it uses a reconciliation loop.

Resync and Reconciliation

The scheduler periodically polls the database for all active triggers. It compares the latest_revision of the triggers in the database with the versions it currently has in memory.

• New Trigger: If a trigger exists in the DB but not in memory, the scheduler starts a new cron job.
• Updated Trigger: If the latest_revision has changed, the scheduler stops the old job and starts a new one with the updated spec.
• Removed/Deactivated Trigger: If a trigger is no longer in the "active" set returned by the DB, the scheduler stops the corresponding job.

Catchup Logic

If the scheduler worker is down or a trigger is temporarily deactivated, executions might be missed. On startup or resync, the scheduler calculates missed runs since the last TriggeredAt or UpdatedAt time.

To prevent a "thundering herd" of executions, this catchup behavior is capped by the MaxCatchupRunsPerLoop configuration setting.

Configuration

You can tune the behavior of the scheduler worker using the TriggerSchedulerConfig in runs/config/config.go.

Parameter	Description
Enabled	Turns the background scheduler worker on or off.
ResyncInterval	How often (e.g., 30s) the worker polls the DB for active triggers.
MaxCatchupRunsPerLoop	Caps the number of missed runs fired in a single loop.
ExecutionQPS	Rate limit (tokens/sec) for the CreateRun calls triggered by the scheduler.
ExecutionBurst	The burst size for the rate limiter.

Example configuration:

{
  "enabled": true,
  "resyncInterval": "30s",
  "maxCatchupRunsPerLoop": 100,
  "executionQps": 10.0,
  "executionBurst": 20
}