Overview of Flyte Caching Architecture

Flyte implements a centralized caching architecture that allows task executions to be memoized, significantly reducing redundant computation and cost. This system coordinates between the execution engine, a specialized cache service, and object storage to ensure that if a task has been run before with the same inputs, its results can be reused immediately.

The Caching Lifecycle

When a task is scheduled for execution, the TaskActionReconciler (located in executor/pkg/controller/taskaction_cache.go) orchestrates the cache interaction. The process follows a strict pre-execution and post-execution flow:

1. Pre-execution Lookup: Before invoking the task plugin, the reconciler calls evaluateCacheBeforeExecution. It constructs a cache key and queries the Catalog client.
2. Cache Hit: If a valid entry is found, the reconciler retrieves the output URIs from the cache service, reads the data from object storage, and populates the task's output writer. The task execution is then short-circuited with a PhaseInfoSuccess.
3. Cache Miss: If no entry exists, the reconciler proceeds to execute the task. If the task is configured for "serializable" caching, it attempts to acquire a reservation (see below).
4. Post-execution Population: Once the task completes successfully, finalizeCacheAfterExecution is called. It invokes writeTaskOutputsToCache, which uploads the task's output metadata to the cache service via the Catalog.Put method.

Cache Key Generation

The uniqueness of a cache entry is determined by its key. The Client in flyteplugins/go/tasks/pluginmachinery/catalog/cache_service/client.go generates this key by combining several factors in the buildCacheKey function:

func buildCacheKey(ctx context.Context, key catalog.Key) (string, error) {
    // ... validation ...
    identifierHash, err := catalog.HashIdentifierExceptVersion(ctx, key.Identifier)
    signatureHash, err := generateInterfaceSignatureHash(ctx, key.TypedInterface)
    inputsHash, err := hashInputs(ctx, key)

    return fmt.Sprintf("%s-%s-%s-%s", identifierHash, signatureHash, inputsHash, key.CacheVersion), nil
}

The key is a composite of:

• Identifier Hash: A hash of the task's project, domain, and name (excluding the version).
• Signature Hash: A hash of the task's input and output variable definitions (the TypedInterface).
• Inputs Hash: A deterministic hash of the actual literal values passed as inputs.
• Cache Version: A user-defined string (from the task definition) that allows manual cache invalidation.

Serializable Caching and Reservations

To prevent the "thundering herd" problem—where multiple concurrent executions of the same task all compute the same result—Flyte uses a reservation system. This is enabled when a task's metadata has CacheSerializable set to true.

The Manager in cache_service/manager/manager.go coordinates these reservations using the GetOrExtendReservation method. When a worker encounters a cache miss for a serializable task, it attempts to become the "owner" of that cache key:

func (m *Manager) GetOrExtendReservation(ctx context.Context, request *cacheservicepb.GetOrExtendReservationRequest, now time.Time) (*cacheservicepb.Reservation, error) {
    // ...
    reservation := &models.Reservation{
        Key:              reservationKey,
        OwnerID:          request.GetOwnerId(),
        HeartbeatSeconds: int64(heartbeat.Seconds()),
        ExpiresAt:        now.Add(heartbeat * time.Duration(m.heartbeatGracePeriodMultiplier)),
    }

    // UpdateIfExpiredOrOwned ensures only one worker can hold the reservation
    if err := m.reservations.UpdateIfExpiredOrOwned(ctx, reservation, now); err != nil {
        // If another worker owns it, the current worker must wait
    }
    // ...
}

If a worker does not own the reservation, the TaskActionReconciler transitions the task to PhaseInfoWaitingForCache. The worker will periodically retry the lookup until the owner populates the cache or the reservation expires.

Heartbeats and Expiration

Reservations are not permanent. The owner must provide heartbeats to maintain its claim. The expiration is calculated as: ExpiresAt = Now + (HeartbeatInterval * HeartbeatGracePeriodMultiplier)

The default HeartbeatGracePeriodMultiplier is 3. If a worker fails or is partitioned, the reservation will eventually expire, allowing another worker to take over and execute the task.

Storage and Metadata Model

The Flyte cache service does not store the actual output data (e.g., large DataFrames or files). Instead, it acts as a metadata index.

• Object Storage (S3/GCS): Stores the actual LiteralMap containing the task outputs.
• Cache Service Database: Stores a mapping from the cache key to the OutputURI (the path in object storage) and execution metadata.

When Manager.Put is called, it records the OutputURI and the LastUpdated timestamp. This separation ensures that the cache service remains lightweight and performant even when handling tasks with massive output data.

Configuration

You can tune the cache service behavior via the following configuration parameters:

• cache_service.heartbeatGracePeriodMultiplier: Controls how many missed heartbeats are allowed before a reservation is considered expired.
• cache_service.maxReservationHeartbeat: The maximum interval a client can request for heartbeats (default 10s).
• catalogCache.maxCacheAge: If set, cache entries older than this duration will be treated as misses, forcing a re-execution.