operator

package
v0.104.0 Latest Latest
Warning

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

 Go to latest Published: Aug 20, 2026 License: MIT Imports: 12 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type DAGStepTriggerRequest 

type DAGStepTriggerRequest struct {
	ParentTaskExternalId uuid.UUID
	InvocationCount      int32
	WorkflowName         string
	// WorkflowVersionId pins triggering to the DAG's original version.
	WorkflowVersionId   uuid.UUID
	ActionId            string
	ChildIndex          int32
	Input               string
	AdditionalMetadata  []byte
	DagParentTaskRunIds []uuid.UUID
	IsSkipped           bool
	IsCancelled         bool
	DesiredWorkerLabels []*sqlcv1.GetDesiredLabelsRow

	// ParentReExecuted forces the step to re-run during a replay when any of its parents
	// re-executed this invocation.
	ParentReExecuted bool
}

type DAGStepTriggerResult 

type DAGStepTriggerResult struct {
	NodeId                int64
	BranchId              int64
	WorkflowRunExternalId uuid.UUID

	IsSatisfied   bool
	ResultPayload []byte
	IsFailure     bool
	ErrorMessage  *string

	// ReExecuted is true when the step actually runs this invocation rather than being
	// satisfied from the log.
	ReExecuted bool
}

type Operator 

type Operator interface {
	HandleAction(ctx context.Context, action *contracts.AssignedAction) error

	WorkerId() uuid.UUID

	// Cleanup tears down a single operator: it pauses the operator's worker and then drains.
	Cleanup()

	// Drain stops accepting new tasks and waits for in-flight ones, without pausing the
	// worker. Used for bulk teardown, where the caller pauses all workers in one query
	// instead of one update per operator.
	Drain()
}

type SharedOperator 

type SharedOperator[T any] struct {
	// contains filtered or unexported fields
}

func NewSharedOperator 

func NewSharedOperator[T any](operator *sqlcv1.V1Operator, l *zerolog.Logger, repo repository.Repository, taskEventWriter TaskEventWriter, workerId uuid.UUID, t T) (*SharedOperator[T], error)

NewSharedOperator constructs the shared operator state.

func (*SharedOperator[T]) CancelDAGChildren 

func (s *SharedOperator[T]) CancelDAGChildren(ctx context.Context, taskExternalIds []uuid.UUID) error

func (*SharedOperator[T]) CancelTask 

func (s *SharedOperator[T]) CancelTask(taskRunExternalId string) bool

func (*SharedOperator[T]) Cleanup 

func (s *SharedOperator[T]) Cleanup()

func (*SharedOperator[T]) Config 

func (s *SharedOperator[T]) Config() T

func (*SharedOperator[T]) Drain 

func (s *SharedOperator[T]) Drain()

func (*SharedOperator[T]) Logger 

func (s *SharedOperator[T]) Logger() *zerolog.Logger

func (*SharedOperator[T]) RecordTask 

func (s *SharedOperator[T]) RecordTask() func()

RecordTask registers an in-flight task and returns a release function that the caller must invoke (typically via defer) when the task finishes. Cleanup blocks until every recorded task has been released.

If the operator is already shutting down, the returned release is a no-op and the task is not tracked — callers should generally avoid starting new work once Cleanup has begun, but in-flight work recorded before shutdown is always awaited.

func (*SharedOperator[T]) RegisterCancellableContext 

func (s *SharedOperator[T]) RegisterCancellableContext(ctx context.Context, taskRunExternalId string) (context.Context, func())

func (*SharedOperator[T]) RegisterDurableTask 

func (s *SharedOperator[T]) RegisterDurableTask(ctx context.Context, externalId uuid.UUID) (chan<- *v1contracts.DurableTaskRequest, <-chan *v1contracts.DurableTaskResponse, error)

RegisterDurableTask opens a channel-based durable-task session through the dispatcher, injecting the tenant the dispatcher reads off the context (the same key sendStepActionEvent uses). Operators that drive durable execution write requests to the returned channel and read responses from it.

func (*SharedOperator[T]) SendCancelled 

func (s *SharedOperator[T]) SendCancelled(action *contracts.AssignedAction) error

SendCancelled reports a cancelled task

func (*SharedOperator[T]) SendCancelledWithMessage 

func (s *SharedOperator[T]) SendCancelledWithMessage(action *contracts.AssignedAction, msg string) error

SendCancelledWithMessage reports a cancelled task with a custom cancellation reason, via the dedicated CancelTaskEvent API rather than the generic step-action-event path.

func (*SharedOperator[T]) SendCompleted 

func (s *SharedOperator[T]) SendCompleted(action *contracts.AssignedAction, output []byte) error

SendCompleted reports a successful result. output should be the task's JSON output.

func (*SharedOperator[T]) SendFailed 

func (s *SharedOperator[T]) SendFailed(action *contracts.AssignedAction, errMsg string, shouldNotRetry bool) error

SendFailed reports a failure with the given error message. shouldNotRetry, when true, prevents the task from being retried.

func (*SharedOperator[T]) SendStarted 

func (s *SharedOperator[T]) SendStarted(action *contracts.AssignedAction) error

SendStarted reports that the operator has started processing the assigned action.

func (*SharedOperator[T]) TenantId 

func (s *SharedOperator[T]) TenantId() uuid.UUID

func (*SharedOperator[T]) TriggerDAGStep 

func (s *SharedOperator[T]) TriggerDAGStep(ctx context.Context, req *DAGStepTriggerRequest) (*DAGStepTriggerResult, error)

func (*SharedOperator[T]) UpdateWorkerActions 

func (s *SharedOperator[T]) UpdateWorkerActions(ctx context.Context, actions []string) error

func (*SharedOperator[T]) WorkerId 

func (s *SharedOperator[T]) WorkerId() uuid.UUID

type TaskEventWriter 

type TaskEventWriter interface {
	SendStepActionEvent(ctx context.Context, request *contracts.StepActionEvent) (*contracts.ActionEventResponse, error)

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

	// RegisterDurableTask opens a channel-based durable-task session: the operator (acting as
	// a durable worker) writes requests to the returned channel and reads responses from it.
	RegisterDurableTask(ctx context.Context, externalId uuid.UUID) (chan<- *v1contracts.DurableTaskRequest, <-chan *v1contracts.DurableTaskResponse, error)

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

	// CancelDAGChildren cancels already-triggered children when the orchestrator is cancelled.
	CancelDAGChildren(ctx context.Context, tenantId uuid.UUID, taskExternalIds []uuid.UUID) error
}

Source Files

View all Source files

Directories

Path Synopsis
safeclient
Package safeclient provides an SSRF-hardened HTTP client for delivering outbound requests to user-supplied endpoints.
 Package safeclient provides an SSRF-hardened HTTP client for delivering outbound requests to user-supplied endpoints.

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.