dispatcher

package
v0.105.21 Latest Latest
Warning

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

 Go to latest Published: Sep 3, 2026 License: MIT Imports: 47 Imported by: 0

Documentation

Index

Constants

View Source 
const HeartbeatInterval = 4 * time.Second

Variables

View Source 
var ErrWorkerNotFound = fmt.Errorf("worker not found")

Functions

func UnmarshalPayload 

func UnmarshalPayload[T any](payload interface{}) (T, error)

Types

type Dispatcher 

type Dispatcher interface {
	contracts.DispatcherServer
	Start() (func() error, error)
}

type DispatcherImpl 

type DispatcherImpl struct {
	contracts.UnimplementedDispatcherServer
	// contains filtered or unexported fields
}

func New 

func New(fs ...DispatcherOpt) (*DispatcherImpl, error)

func (*DispatcherImpl) CancelDAGChildren added in v0.104.0 

func (d *DispatcherImpl) CancelDAGChildren(ctx context.Context, tenantId uuid.UUID, taskExternalIds []uuid.UUID) error

func (*DispatcherImpl) CancelStreamSessions added in v0.89.2 

func (d *DispatcherImpl) CancelStreamSessions()

CancelStreamSessions hangs up all registered long-lived subscriber streams. It is called during shutdown before GracefulStop, which would otherwise block on them until the process is killed.

func (*DispatcherImpl) CancelTaskEvent added in v0.104.0 

func (d *DispatcherImpl) CancelTaskEvent(ctx context.Context, request *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)

func (*DispatcherImpl) DispatcherId added in v0.80.0 

func (d *DispatcherImpl) DispatcherId() uuid.UUID

func (*DispatcherImpl) GetLocalWorkerIds added in v0.78.0 

func (d *DispatcherImpl) GetLocalWorkerIds() map[uuid.UUID]struct{}

func (*DispatcherImpl) GetVersion added in v0.78.27 

func (s *DispatcherImpl) GetVersion(ctx context.Context, req *contracts.GetVersionRequest) (*contracts.GetVersionResponse, error)

func (*DispatcherImpl) HandleLocalAssignments added in v0.78.0 

func (d *DispatcherImpl) HandleLocalAssignments(ctx context.Context, tenantId, workerId uuid.UUID, tasks []*schedulingv1.AssignedItemWithTask) error

Note: this is very similar to handleTaskBulkAssignedTask, with some differences in what's sync vs run in a goroutine In this method, we wait until all tasks have been sent to the worker before returning

func (*DispatcherImpl) Heartbeat 

func (s *DispatcherImpl) Heartbeat(ctx context.Context, req *contracts.HeartbeatRequest) (*contracts.HeartbeatResponse, error)

Heartbeat is used to update the last heartbeat time for a worker

func (*DispatcherImpl) Listen 

func (s *DispatcherImpl) Listen(request *contracts.WorkerListenRequest, stream contracts.Dispatcher_ListenServer) error

Subscribe handles a subscribe request from a client

func (*DispatcherImpl) ListenV2 

func (s *DispatcherImpl) ListenV2(request *contracts.WorkerListenRequest, stream contracts.Dispatcher_ListenV2Server) error

ListenV2 is like Listen, but implementation does not include heartbeats. This should only used by SDKs against engine version v0.18.1+

func (*DispatcherImpl) PutOverridesData 

func (s *DispatcherImpl) PutOverridesData(ctx context.Context, request *contracts.OverridesData) (*contracts.OverridesDataResponse, error)

func (*DispatcherImpl) RefreshTimeout 

func (d *DispatcherImpl) RefreshTimeout(ctx context.Context, request *contracts.RefreshTimeoutRequest) (*contracts.RefreshTimeoutResponse, error)

func (*DispatcherImpl) Register 

func (s *DispatcherImpl) Register(ctx context.Context, request *contracts.WorkerRegisterRequest) (*contracts.WorkerRegisterResponse, error)

func (*DispatcherImpl) RegisterDurableTask added in v0.104.0 

func (d *DispatcherImpl) RegisterDurableTask(ctx context.Context, externalId uuid.UUID) (chan<- *contracts.DurableTaskRequest, <-chan *contracts.DurableTaskResponse, error)

RegisterDurableTask delegates to the V1 dispatcher service so DispatcherImpl satisfies operator.TaskEventWriter, giving operators the in-engine equivalent of the DurableTask RPC.

func (*DispatcherImpl) ReleaseSlot 

func (s *DispatcherImpl) ReleaseSlot(ctx context.Context, req *contracts.ReleaseSlotRequest) (*contracts.ReleaseSlotResponse, error)

func (*DispatcherImpl) RestoreEvictedTask added in v0.80.0 

func (s *DispatcherImpl) RestoreEvictedTask(ctx context.Context, req *contracts.RestoreEvictedTaskRequest) (*contracts.RestoreEvictedTaskResponse, error)

func (*DispatcherImpl) SendBatchActionEvent added in v0.98.0 

func (s *DispatcherImpl) SendBatchActionEvent(ctx context.Context, request *contracts.BatchActionEvent) (*contracts.ActionEventResponse, error)

func (*DispatcherImpl) SendGroupKeyActionEvent 

func (s *DispatcherImpl) SendGroupKeyActionEvent(ctx context.Context, request *contracts.GroupKeyActionEvent) (*contracts.ActionEventResponse, error)

func (*DispatcherImpl) SendStepActionEvent 

func (s *DispatcherImpl) SendStepActionEvent(ctx context.Context, request *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)

func (*DispatcherImpl) Start 

func (d *DispatcherImpl) Start() (func() error, error)

func (*DispatcherImpl) SubscribeToWorkflowEvents 

func (s *DispatcherImpl) SubscribeToWorkflowEvents(request *contracts.SubscribeToWorkflowEventsRequest, stream contracts.Dispatcher_SubscribeToWorkflowEventsServer) error

func (*DispatcherImpl) SubscribeToWorkflowRuns 

func (s *DispatcherImpl) SubscribeToWorkflowRuns(server contracts.Dispatcher_SubscribeToWorkflowRunsServer) error

func (*DispatcherImpl) TriggerDAGStep added in v0.104.0 

func (d *DispatcherImpl) TriggerDAGStep(ctx context.Context, tenantId uuid.UUID, req *operator.DAGStepTriggerRequest) (*operator.DAGStepTriggerResult, error)

func (*DispatcherImpl) Unsubscribe 

func (s *DispatcherImpl) Unsubscribe(ctx context.Context, request *contracts.WorkerUnsubscribeRequest) (*contracts.WorkerUnsubscribeResponse, error)

func (*DispatcherImpl) UpsertWorkerLabels 

func (s *DispatcherImpl) UpsertWorkerLabels(ctx context.Context, request *contracts.UpsertWorkerLabelsRequest) (*contracts.UpsertWorkerLabelsResponse, error)

func (*DispatcherImpl) V1 added in v0.89.3 

func (d *DispatcherImpl) V1() *DispatcherServiceImpl

V1 returns the dispatcher's V1Dispatcher gRPC service, which serves the durable task and durable event RPCs.

type DispatcherOpt 

type DispatcherOpt func(*DispatcherOpts)

func WithAlerter 

func WithAlerter(a hatcheterrors.Alerter) DispatcherOpt

func WithAnalytics added in v0.79.36 

func WithAnalytics(a analytics.Analytics) DispatcherOpt

func WithCache 

func WithCache(cache cache.Cacheable) DispatcherOpt

func WithDAGOperatorDefaultSlots added in v0.104.1 

func WithDAGOperatorDefaultSlots(slots int) DispatcherOpt

func WithDataDecoderValidator 

func WithDataDecoderValidator(dv datautils.DataDecoderValidator) DispatcherOpt

func WithDefaultMaxWorkerLockAcquisitionTime added in v0.80.5 

func WithDefaultMaxWorkerLockAcquisitionTime(t time.Duration) DispatcherOpt

func WithDispatcherId 

func WithDispatcherId(dispatcherId uuid.UUID) DispatcherOpt

func WithEncryption added in v0.104.0 

func WithEncryption(enc encryption.EncryptionService) DispatcherOpt

func WithInfraBlockedCIDRs added in v0.104.0 

func WithInfraBlockedCIDRs(cidrs []string) DispatcherOpt

func WithLogger 

func WithLogger(l *zerolog.Logger) DispatcherOpt

func WithMessageQueueV1 

func WithMessageQueueV1(mqv1 msgqueue.MessageQueue) DispatcherOpt

func WithPayloadSizeThreshold 

func WithPayloadSizeThreshold(threshold int) DispatcherOpt

func WithPrometheusGate added in v0.90.0 

func WithPrometheusGate(gate *prometheus.Gate) DispatcherOpt

func WithPubSub added in v0.98.9 

func WithPubSub(pubsub msgqueue.PubSub) DispatcherOpt

func WithRepositoryV1 

func WithRepositoryV1(r v1.Repository) DispatcherOpt

func WithStreamEventBufferTimeout added in v0.79.34 

func WithStreamEventBufferTimeout(timeout time.Duration) DispatcherOpt

func WithVersion added in v0.78.27 

func WithVersion(version string) DispatcherOpt

func WithWorkflowRunBufferSize added in v0.78.0 

func WithWorkflowRunBufferSize(size int) DispatcherOpt

type DispatcherOpts 

type DispatcherOpts struct {
	// contains filtered or unexported fields
}

type DispatcherServiceImpl added in v0.89.3 

type DispatcherServiceImpl struct {
	contracts.UnimplementedV1DispatcherServer
	// contains filtered or unexported fields
}

func (*DispatcherServiceImpl) CancelDAGChildren added in v0.104.0 

func (d *DispatcherServiceImpl) CancelDAGChildren(ctx context.Context, tenantId uuid.UUID, taskExternalIds []uuid.UUID) error

CancelDAGChildren cancels already-triggered DAG children on the operator's behalf, since it has no direct message-queue access; mirrors the admin CancelTasks path.

func (*DispatcherServiceImpl) CancelStreamSessions added in v0.89.3 

func (d *DispatcherServiceImpl) CancelStreamSessions()

CancelStreamSessions hangs up all registered long-lived streams (durable event and durable task listeners). It is called during shutdown before GracefulStop, which would otherwise block on them until the process is killed.

func (*DispatcherServiceImpl) DeliverDurableEventLogEntryCompletion added in v0.89.3 

func (d *DispatcherServiceImpl) DeliverDurableEventLogEntryCompletion(tenantId uuid.UUID, taskExternalId uuid.UUID, invocationCount int32, branchId, nodeId int64, payload []byte, satisfiedOrder *int64, isFailure bool, errorMessage *string) error

func (*DispatcherServiceImpl) DurableTask added in v0.89.3 

func (d *DispatcherServiceImpl) DurableTask(server contracts.V1Dispatcher_DurableTaskServer) error

func (*DispatcherServiceImpl) ListenForDurableEvent added in v0.89.3 

func (d *DispatcherServiceImpl) ListenForDurableEvent(server contracts.V1Dispatcher_ListenForDurableEventServer) error

func (*DispatcherServiceImpl) RegisterDurableEvent added in v0.89.3 

func (d *DispatcherServiceImpl) RegisterDurableEvent(ctx context.Context, req *contracts.RegisterDurableEventRequest) (*contracts.RegisterDurableEventResponse, error)

func (*DispatcherServiceImpl) RegisterDurableTask added in v0.104.0 

func (d *DispatcherServiceImpl) RegisterDurableTask(ctx context.Context, externalId uuid.UUID) (chan<- *contracts.DurableTaskRequest, <-chan *contracts.DurableTaskResponse, error)

RegisterDurableTask sets up a channel-backed durable-task session — the in-engine equivalent of the DurableTask gRPC stream, for operators that don't hold a gRPC stream. The caller writes DurableTaskRequests to the returned requestCh and reads DurableTaskResponses from respCh, reusing the same handlers and routing table (durableInvocations) as the gRPC path, so async responses (wait-for satisfied, evictions, acks) are delivered identically.

externalId is registered up front so responses route immediately; additional task ids are registered lazily as messages reference them, matching DurableTask. The session is torn down (invocations deregistered, respCh closed) when ctx is cancelled or requestCh is closed.

func (*DispatcherServiceImpl) TriggerDAGStep added in v0.104.0 

func (d *DispatcherServiceImpl) TriggerDAGStep(ctx context.Context, tenantId uuid.UUID, req *operator.DAGStepTriggerRequest) (*operator.DAGStepTriggerResult, error)

type StreamEventBuffer 

type StreamEventBuffer struct {
	// contains filtered or unexported fields
}

func NewStreamEventBuffer 

func NewStreamEventBuffer(timeout time.Duration) *StreamEventBuffer

func (*StreamEventBuffer) AddEvent 

func (b *StreamEventBuffer) AddEvent(event *contracts.WorkflowEvent)

func (*StreamEventBuffer) Close 

func (b *StreamEventBuffer) Close()

Close stops the buffer's background goroutines. The channels are deliberately not closed: in-flight AddEvent/timeout sends race a close (send on a closed channel panics even inside a select), and every sender and consumer already exits via context cancellation.

func (*StreamEventBuffer) Events 

func (b *StreamEventBuffer) Events() <-chan *contracts.WorkflowEvent

type V1TaskWithPayloadAndInvocationCount added in v0.80.0 

type V1TaskWithPayloadAndInvocationCount struct {
	*v1.V1TaskWithPayload
	InvocationCount *int32 // only used for durable tasks
}

Source Files

View all Source files

Directories

Path Synopsis

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.