dispatch

package
v2.15.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 22, 2026 License: GPL-3.0 Imports: 6 Imported by: 0

Documentation

Overview

Package dispatch holds control-plane policy for deciding how a DAG run is executed. It is platform code: it combines language-level DAG fields with runtime configuration to choose between local and coordinator-dispatched execution.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrDispatchTaskNotFound                   = errors.New("dispatch task claim not found")
	ErrDispatchAdmissionNotFound              = errors.New("dispatch admission not found")
	ErrDispatchAdmissionConflict              = errors.New("dispatch admission conflict")
	ErrDispatchAdmissionLivenessNotConfigured = errors.New("dispatch admission liveness not configured")
	ErrDAGRunLeaseNotFound                    = errors.New("dag-run lease not found")
	ErrDAGRunLeaseConflict                    = errors.New("dag-run lease identity conflict")
	ErrActiveRunNotFound                      = errors.New("active distributed run not found")
	ErrWorkerHeartbeatNotFound                = errors.New("worker heartbeat not found")
)
View Source 
var ErrBuildRequiresLocal = errors.New("build workflows require local execution; distributed fencing is not implemented")

ErrBuildRequiresLocal reports that distributed materialization fencing is unavailable.

Functions

func AttemptKeyForStatus added in v2.13.0 

func AttemptKeyForStatus(status *ir.DAGRunStatus, fallbackAttemptID string) string

AttemptKeyForStatus resolves the authoritative attempt key for a persisted status. When older statuses omit AttemptKey, it is regenerated from the stored DAG-run identity and attempt ID.

func DistributedLeaseExpiredReason added in v2.13.0 

func DistributedLeaseExpiredReason(workerID string) string

DistributedLeaseExpiredReason is the canonical failure reason for a distributed attempt that lost authoritative worker ownership.

func IsRemoteWorkerID added in v2.13.0 

func IsRemoteWorkerID(workerID string) bool

IsRemoteWorkerID reports whether the status originated from a distributed worker instead of the local process runtime.

func LeaseIdentityMatchesStatus added in v2.13.0 

func LeaseIdentityMatchesStatus(
	lease *DAGRunLease,
	status *ir.DAGRunStatus,
	fallbackAttemptID string,
) bool

LeaseIdentityMatchesStatus reports whether the lease belongs to the same persisted distributed attempt as status, independent of freshness.

func LeaseMatchesStatus added in v2.13.0 

func LeaseMatchesStatus(
	lease *DAGRunLease,
	status *ir.DAGRunStatus,
	fallbackAttemptID string,
	now time.Time,
	staleThreshold time.Duration,
) bool

LeaseMatchesStatus reports whether the lease still authoritatively belongs to the exact distributed attempt represented by the persisted status.

func ShouldDispatchToCoordinator 

func ShouldDispatchToCoordinator(dag *ir.DAG, hasCoordinator bool, defaultMode config.ExecutionMode) bool

ShouldDispatchToCoordinator decides whether a DAG should be dispatched to the coordinator for distributed execution.

Types

type ActiveDistributedRun added in v2.13.0 

type ActiveDistributedRun struct {
	AttemptKey string       `json:"attemptKey"`
	DAGRun     ir.DAGRunRef `json:"dagRun"`
	Root       ir.DAGRunRef `json:"root,omitzero"`
	AttemptID  string       `json:"attemptId"`
	WorkerID   string       `json:"workerId"`
	Status     ir.Status    `json:"status"`
	UpdatedAt  int64        `json:"updatedAt"`
}

ActiveDistributedRun is the durable active-set record for a remote attempt.

type ActiveDistributedRunStore added in v2.13.0 

type ActiveDistributedRunStore interface {
	Upsert(ctx context.Context, record ActiveDistributedRun) error
	Delete(ctx context.Context, attemptKey string) error
	Get(ctx context.Context, attemptKey string) (*ActiveDistributedRun, error)
	ListAll(ctx context.Context) ([]ActiveDistributedRun, error)
}

ActiveDistributedRunStore persists the coordinator-owned active distributed attempt index used by the zombie detector.

type ClaimedDispatchTask added in v2.13.0 

type ClaimedDispatchTask struct {
	Task       *DispatchTask
	ClaimToken string
	ClaimedAt  time.Time
	WorkerID   string
	PollerID   string
	Owner      CoordinatorEndpoint
}

ClaimedDispatchTask is a shared pending task that has been claimed by a specific worker poller and must be acknowledged before execution begins.

type CoordinatorEndpoint added in v2.13.0 

type CoordinatorEndpoint struct {
	ID   string `json:"id,omitempty"`
	Host string `json:"host,omitempty"`
	Port int    `json:"port,omitempty"`
}

CoordinatorEndpoint identifies a coordinator instance that owns a distributed task or run.

type DAGRunLease added in v2.13.0 

type DAGRunLease struct {
	AttemptKey      string              `json:"attemptKey"`
	DAGRun          ir.DAGRunRef        `json:"dagRun"`
	Root            ir.DAGRunRef        `json:"root,omitzero"`
	AttemptID       string              `json:"attemptId"`
	QueueName       string              `json:"queueName"`
	WorkerID        string              `json:"workerId"`
	Owner           CoordinatorEndpoint `json:"owner"`
	ClaimToken      string              `json:"claimToken,omitempty"`
	ClaimedAt       int64               `json:"claimedAt"`
	LastHeartbeatAt int64               `json:"lastHeartbeatAt"`
}

DAGRunLease is the shared liveness record for an accepted worker claim.

func (DAGRunLease) IsFresh added in v2.13.0 

func (l DAGRunLease) IsFresh(now time.Time, staleThreshold time.Duration) bool

IsFresh reports whether the lease is still alive.

func (DAGRunLease) LastHeartbeatTime added in v2.13.0 

func (l DAGRunLease) LastHeartbeatTime() time.Time

LastHeartbeatTime returns the last run heartbeat time.

func (*DAGRunLease) MatchesClaim added in v2.13.0 

func (l *DAGRunLease) MatchesClaim(claimKey, workerID string) bool

MatchesClaim reports whether the lease identifies the worker claim.

type DAGRunLeaseStore added in v2.13.0 

type DAGRunLeaseStore interface {
	Upsert(ctx context.Context, lease DAGRunLease) error
	Touch(ctx context.Context, attemptKey string, observedAt time.Time) error
	Delete(ctx context.Context, attemptKey string) error
	Get(ctx context.Context, attemptKey string) (*DAGRunLease, error)
	ListByQueue(ctx context.Context, queueName string) ([]DAGRunLease, error)
	ListAll(ctx context.Context) ([]DAGRunLease, error)
}

DAGRunLeaseStore persists live worker-claim leases.

type DAGRunStatusResult added in v2.13.0 

type DAGRunStatusResult struct {
	Found  bool
	Status *ir.DAGRunStatus
}

DAGRunStatusResult is a distributed status lookup result.

type DispatchAdmissionBindRequest added in v2.13.0 

type DispatchAdmissionBindRequest struct {
	ReservationToken string
	Task             *DispatchTask
}

DispatchAdmissionBindRequest describes a coordinator bind for a reservation.

type DispatchAdmissionDecision added in v2.13.0 

type DispatchAdmissionDecision struct {
	Reserved         bool
	Reason           DispatchAdmissionRejectReason
	ReservationToken string
}

DispatchAdmissionDecision is the durable queue admission result.

type DispatchAdmissionRejectReason added in v2.13.0 

type DispatchAdmissionRejectReason string

DispatchAdmissionRejectReason identifies why a reservation was not granted. 

const (
	DispatchAdmissionRejectedDuplicate  DispatchAdmissionRejectReason = "duplicate"
	DispatchAdmissionRejectedNoCapacity DispatchAdmissionRejectReason = "no_capacity"
)

type DispatchAdmissionRequest added in v2.13.0 

type DispatchAdmissionRequest struct {
	QueueName             string
	MaxConcurrency        int
	NonAdmissionOccupancy int
	AttemptKey            string
	AttemptID             string
	DAGRun                ir.DAGRunRef
	StaleThreshold        time.Duration
}

DispatchAdmissionRequest describes a queue admission reservation request.

type DispatchAdmissionStore added in v2.13.0 

type DispatchAdmissionStore interface {
	ReserveAdmission(ctx context.Context, req DispatchAdmissionRequest) (*DispatchAdmissionDecision, error)
	BindAdmission(ctx context.Context, req DispatchAdmissionBindRequest) error
	ReleaseAdmissionToken(ctx context.Context, reservationToken string) error
	FinalizeAdmissionAttempt(ctx context.Context, attemptKey string) error
	CleanupAdmissions(ctx context.Context, staleThreshold time.Duration) error
}

DispatchAdmissionStore reserves queue capacity and binds reservations to tasks.

type DispatchOperation added in v2.13.0 

type DispatchOperation int32

DispatchOperation identifies the operation requested for a distributed DAG run. 

const (
	DispatchOperationUnspecified DispatchOperation = iota
	DispatchOperationStart
	DispatchOperationRetry
)

func (DispatchOperation) String added in v2.13.0 

func (o DispatchOperation) String() string

type DispatchRequest added in v2.13.0 

type DispatchRequest struct {
	Task                      *DispatchTask
	AdmissionReservationToken string
}

DispatchRequest describes a distributed dispatch call.

type DispatchTask added in v2.13.0 

type DispatchTask struct {
	RootDAGRunName string
	RootDAGRunID   string

	ParentDAGRunName string
	ParentDAGRunID   string

	Operation      DispatchOperation
	DAGRunID       string
	Target         string
	Definition     string
	AttemptID      string
	AttemptKey     string
	Step           string
	Params         string
	ParallelItem   string
	QueueName      string
	WorkerID       string
	TargetWorkerID string
	ProfileName    string
	DefinitionID   string
	TriggerActor   string

	PreviousStatus *ir.DAGRunStatus

	BaseConfig   string
	Labels       string
	ScheduleTime string
	SourceFile   string

	WorkerSelector map[string]string

	ExternalStepRetry bool
	IncludeDownstream bool
	RetryPath         string

	WorkspaceBundleDigest      string
	WorkspaceBundleSize        int64
	WorkspaceBundleDAGPath     string
	WorkspaceBundleOriginalRef string
	WorkspaceBundleResolvedRef string

	Owner      CoordinatorEndpoint
	ClaimToken string
}

DispatchTask describes a DAG run request for a distributed executor.

type DispatchTaskClaim added in v2.13.0 

type DispatchTaskClaim struct {
	WorkerID     string
	PollerID     string
	Labels       map[string]string
	Owner        CoordinatorEndpoint
	ClaimTimeout time.Duration
}

DispatchTaskClaim describes a worker poller's attempt to claim a shared distributed task.

type DispatchTaskStore added in v2.13.0 

type DispatchTaskStore interface {
	Enqueue(ctx context.Context, task *DispatchTask) error
	ClaimNext(ctx context.Context, claim DispatchTaskClaim) (*ClaimedDispatchTask, error)
	GetClaim(ctx context.Context, claimToken string) (*ClaimedDispatchTask, error)
	ReleaseClaim(ctx context.Context, claimToken string) error
	DeleteClaim(ctx context.Context, claimToken string) error
	CountOutstandingByQueue(ctx context.Context, queueName string, claimTimeout time.Duration) (int, error)
	HasOutstandingAttempt(ctx context.Context, attemptKey string, claimTimeout time.Duration) (bool, error)
}

DispatchTaskStore manages the shared distributed dispatch queue.

type Dispatcher added in v2.13.0 

type Dispatcher interface {
	Dispatch(ctx context.Context, req DispatchRequest) error
	Cleanup(ctx context.Context) error
	GetDAGRunStatus(ctx context.Context, dagName, dagRunID string, rootRef *ir.DAGRunRef) (*DAGRunStatusResult, error)
	RequestCancel(ctx context.Context, dagName, dagRunID string, rootRef *ir.DAGRunRef) error
}

Dispatcher defines distributed DAG run operations.

type RunningTask added in v2.13.0 

type RunningTask struct {
	DAGRunID         string `json:"dagRunId,omitempty"`
	DAGName          string `json:"dagName,omitempty"`
	StartedAt        int64  `json:"startedAt,omitempty"`
	RootDAGRunName   string `json:"rootDagRunName,omitempty"`
	RootDAGRunID     string `json:"rootDagRunId,omitempty"`
	ParentDAGRunName string `json:"parentDagRunName,omitempty"`
	ParentDAGRunID   string `json:"parentDagRunId,omitempty"`
	AttemptKey       string `json:"attemptKey,omitempty"`
}

RunningTask describes one task currently executing on a worker.

type WorkerHeartbeatRecord added in v2.13.0 

type WorkerHeartbeatRecord struct {
	WorkerID        string            `json:"workerId"`
	Labels          map[string]string `json:"labels,omitempty"`
	Stats           *WorkerStats      `json:"stats,omitempty"`
	LastHeartbeatAt int64             `json:"lastHeartbeatAt"`
}

WorkerHeartbeatRecord is the shared presence record for a worker.

func (WorkerHeartbeatRecord) LastHeartbeatTime added in v2.13.0 

func (r WorkerHeartbeatRecord) LastHeartbeatTime() time.Time

LastHeartbeatTime returns the last heartbeat as a time.

type WorkerHeartbeatStore added in v2.13.0 

type WorkerHeartbeatStore interface {
	Upsert(ctx context.Context, record WorkerHeartbeatRecord) error
	Get(ctx context.Context, workerID string) (*WorkerHeartbeatRecord, error)
	List(ctx context.Context) ([]WorkerHeartbeatRecord, error)
	DeleteStale(ctx context.Context, before time.Time) (int, error)
}

WorkerHeartbeatStore persists shared worker presence across coordinators.

type WorkerStats added in v2.13.0 

type WorkerStats struct {
	TotalPollers int32          `json:"totalPollers,omitempty"`
	BusyPollers  int32          `json:"busyPollers,omitempty"`
	RunningTasks []*RunningTask `json:"runningTasks,omitempty"`
}

WorkerStats describes worker poller capacity and running distributed tasks.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.