Task and Version Management

Flyte manages tasks through a versioned registration system that ensures execution reproducibility. Tasks are defined by a unique combination of project, domain, name, and version, and their specifications are stored as content-addressable blobs.

Registering and Deploying Tasks

To register a new task or update an existing version, use the DeployTask method in the taskService. This operation is an upsert: if a task with the same identifier already exists, its metadata and specification are updated.

import (
    "context"
    "github.com/flyteorg/flyte/v2/gen/go/flyteidl2/task"
    "github.com/flyteorg/flyte/v2/runs/service"
)

func RegisterTask(ctx context.Context, taskSvc taskconnect.TaskServiceHandler) error {
    req := &task.DeployTaskRequest{
        TaskId: &task.TaskIdentifier{
            Org:     "my-org",
            Project: "my-project",
            Domain:  "development",
            Name:    "prod.my_function", // Convention: environment.function_name
            Version: "v1.2.3",
        },
        Spec: &task.TaskSpec{
            Environment: &task.Environment{
                Name:        "prod",
                Description: "Production environment",
            },
            Documentation: &task.TaskDocumentation{
                ShortDescription: "Processes data records",
            },
        },
    }

    _, err := taskSvc.DeployTask(ctx, connect.NewRequest(req))
    return err
}

When DeployTask is called, Flyte performs the following:

1. Validation: Ensures the project exists and validates trigger configurations (e.g., cron expressions).
2. Metadata Extraction: The ExtractFunctionName utility in runs/repository/transformers/task.go attempts to strip the environment prefix from the task name to identify the underlying function.
3. Persistence: The TaskRepo.CreateTask method executes a single transaction to upsert the Task model and synchronize its associated triggers.

Versioning and Reproducibility

Flyte uses TaskSpec to ensure that workflow executions are reproducible. While the tasks table stores the specification for each version, the system also supports content-addressable storage via the TaskSpec model in runs/repository/models/task_spec.go.

Task Specification Deduplication

The TaskSpec model uses a Base64 encoded digest as a unique identifier. This allows Flyte to deduplicate identical specifications across different versions or tasks.

// From runs/repository/models/task_spec.go
type TaskSpec struct {
    Digest    string    `db:"digest"`    // Unique identifier (hash)
    CreatedAt time.Time `db:"created_at"`
    Spec      []byte    `db:"spec"`      // Marshaled task specification
}

Listing Task Versions

To retrieve the history of a specific task, use ListVersions. This returns a list of TaskVersion models containing the version string and the deployment timestamp.

func GetTaskHistory(ctx context.Context, taskSvc taskconnect.TaskServiceHandler) ([]*task.ListVersionsResponse_VersionResponse, error) {
    req := &task.ListVersionsRequest{
        TaskName: &task.TaskName{
            Project: "my-project",
            Domain:  "development",
            Name:    "prod.my_function",
        },
    }
    
    resp, err := taskSvc.ListVersions(ctx, connect.NewRequest(req))
    if err != nil {
        return nil, err
    }
    return resp.Msg.Versions, nil
}

Managing Task Triggers

Triggers are managed as part of the task deployment lifecycle. When you deploy a task with a list of TaskTrigger objects, Flyte synchronizes the database state to match the request.

• Upsert: New or modified triggers are added or updated.
• Pruning: Any existing triggers for the task that are not included in the DeployTaskRequest are soft-deleted.
• Revisioning: Each trigger update creates a new revision, tracked in the repository layer.

The taskService recomputes trigger metadata (like TotalTriggers and ActiveTriggers) automatically after the transaction commits.

Querying the Latest Tasks

The ListTasks API is designed to provide a high-level view of available tasks. By default, it uses a ROW_NUMBER() window function in the tasksRepo.ListTasks implementation to return only the latest version of each unique task (partitioned by project, domain, and name).

-- Logic used in runs/repository/impl/task.go
WITH filtered_tasks AS (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY project, domain, name ORDER BY created_at DESC) AS rn
    FROM tasks
)
SELECT * FROM filtered_tasks WHERE rn = 1;

Troubleshooting and Constraints

Description Truncation

Flyte enforces limits on description lengths to maintain database performance. The taskService uses internal constants to truncate fields:

• ShortDescription: Truncated if it exceeds the maximum allowed length.
• LongDescription: Truncated if it exceeds the maximum allowed length.

Task Name Conventions

The system expects task names to follow the pattern {environment}.{function_name}. If the Environment field in the TaskSpec is set, Flyte will attempt to strip that name as a prefix. If the prefix does not match, it falls back to using the last segment of the name (split by .) as the function name.

Trigger Validation

If a DeployTask request contains a trigger with an invalid cron expression, the entire deployment transaction will fail before any changes are written to the database. This ensures that tasks are never left in a partially configured state.