The Cache Manager and Service Layer

The Flyte Cache Service is structured into two primary layers: a thin service layer that handles transport and request validation, and a manager layer that implements the core business logic for cache persistence and distributed coordination.

Architecture Overview

The Cache Service follows a layered architecture to separate API concerns from business logic:

1. Service Layer (CacheService): Acts as the entry point for ConnectRPC requests. It validates incoming messages and translates multi-tenant identifiers (Project/Domain) into internal scoped keys.
2. Manager Layer (Manager): Orchestrates cache operations. It manages the lifecycle of cache entries and coordinates "reservations" to prevent multiple workers from computing the same result simultaneously.
3. Repository Layer: Handles the actual persistence of cache metadata and reservations to a database (e.g., PostgreSQL).

The Service Layer

When you send a request to the Cache Service, the CacheService class in cache_service/service/service.go is responsible for the initial handling. It uses the validatableRequest interface to ensure that all incoming protobuf messages are valid before processing.

Key Scoping

Flyte ensures that cache entries are isolated by project and domain. The service layer automatically transforms a user-provided key into a "scoped key" using the scopedKey function:

func scopedKey(key string, id *cacheservicev2.Identifier) string {
	return fmt.Sprintf("%s-%s-%s", id.GetProject(), id.GetDomain(), key)
}

This means if you request a cache entry for key my-task-output in project flytesnacks and domain development, the internal key used by the Manager will be flytesnacks-development-my-task-output.

Request Validation

The service layer enforces strict validation on all requests. It defines a validatableRequest interface that matches the Validate() method generated by protoc-gen-validate:

type validatableRequest interface {
	Validate() error
}

func validateRequest(msg validatableRequest) error {
	if err := msg.Validate(); err != nil {
		return connect.NewError(connect.CodeInvalidArgument, err)
	}
	return nil
}

The Manager Layer

The Manager class in cache_service/manager/manager.go owns the core behavior of the service. It does not store the actual data blobs (which remain in object storage); instead, it stores the OutputURI and associated Metadata.

Retrieving and Storing Entries

When the Manager retrieves an entry, it returns a CacheEntry struct:

type CacheEntry struct {
	OutputURI string
	Metadata  *cacheservicepb.Metadata
}

The Put operation handles metadata merging. If an entry already exists, the Manager ensures that the CreatedAt timestamp is preserved while updating the LastUpdatedAt field via the mergeMetadata helper:

func mergeMetadata(existing *models.CachedOutput, request *cacheservicepb.Metadata, now time.Time) *cacheservicepb.Metadata {
    // ... logic to preserve CreatedAt from existing record ...
	metadata.LastUpdatedAt = timestamppb.New(now)
	return metadata
}

Distributed Coordination with Reservations

To prevent a "thundering herd" problem—where multiple workers attempt to compute and cache the same result at once—Flyte uses a reservation system.

How Reservations Work

When a worker experiences a cache miss, it attempts to "reserve" the right to populate that cache entry. The Manager.GetOrExtendReservation method coordinates this:

1. Claiming: A worker requests a reservation with an OwnerID.
2. Heartbeating: The reservation is valid for a specific duration. The worker must periodically "extend" the reservation to keep it.
3. Exclusion: If another worker tries to claim the same key while a valid reservation exists, the Manager returns the current owner's information instead of granting a new reservation.

func (m *Manager) GetOrExtendReservation(ctx context.Context, request *cacheservicepb.GetOrExtendReservationRequest, now time.Time) (*cacheservicepb.Reservation, error) {
    // ...
	reservationKey := fmt.Sprintf("%s:%s", reservationPrefix, request.GetKey())
    // ...
	if err == nil {
		if err := m.reservations.UpdateIfExpiredOrOwned(ctx, reservation, now); err != nil {
			if repositoryerrors.IsReservationNotClaimable(err) {
				// Another caller still owns the reservation
				current, getErr := m.reservations.Get(ctx, reservationKey)
				return reservationFromModel(current), nil
			}
		}
	}
    // ...
}

Reservations are stored with a reservation: prefix to distinguish them from the actual cache entries.

Configuration

The behavior of the Manager is tuned via the cacheconfig.Config struct. Key parameters include:

• HeartbeatGracePeriodMultiplier: (Default: 3) The number of heartbeat intervals that can pass before a reservation is considered expired. For example, if the heartbeat interval is 10s and the multiplier is 3, the reservation expires after 30s of inactivity.
• MaxReservationHeartbeatInterval: (Default: 10s) The maximum duration a worker can request for a single heartbeat interval.

These values are initialized in the Manager constructor:

func New(cfg *cacheconfig.Config, outputs interfaces.CachedOutputRepo, reservations interfaces.ReservationRepo) *Manager {
	maxHeartbeat := cfg.MaxReservationHeartbeat.Duration
	if maxHeartbeat <= 0 {
		maxHeartbeat = 10 * time.Second
	}

	graceMultiplier := cfg.HeartbeatGracePeriodMultiplier
	if graceMultiplier <= 0 {
		graceMultiplier = 3
	}

	return &Manager{
		outputs:                         outputs,
		reservations:                    reservations,
		heartbeatGracePeriodMultiplier:  graceMultiplier,
		maxReservationHeartbeatInterval: maxHeartbeat,
	}
}