client

package
v1.0.0-beta.1 Latest Latest
Warning

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

 Go to latest Published: Sep 14, 2026 License: Apache-2.0 Imports: 35 Imported by: 7

Documentation

Overview

Package client implements the transport-neutral Durable Task gRPC client and worker used by the Durable Task Scheduler integration.

Index

Constants

View Source 
const (
	WorkerCapabilityHistoryStreaming WorkerCapability = protos.WorkerCapability_WORKER_CAPABILITY_HISTORY_STREAMING
	WorkerCapabilityScheduledTasks   WorkerCapability = protos.WorkerCapability_WORKER_CAPABILITY_SCHEDULED_TASKS
	WorkerCapabilityLargePayloads    WorkerCapability = protos.WorkerCapability_WORKER_CAPABILITY_LARGE_PAYLOADS

	// DefaultMaxStreamedHistoryEvents and DefaultMaxStreamedHistoryBytes are
	// the worker's default per-work-item replay-history safety bounds.
	DefaultMaxStreamedHistoryEvents       = api.DefaultHistoryMaxEvents
	DefaultMaxStreamedHistoryBytes  int64 = api.DefaultHistoryMaxBytes
	// MaxStreamedHistoryEvents and MaxStreamedHistoryBytes bound configurable
	// worker replay-history limits.
	MaxStreamedHistoryEvents       = api.MaxHistoryMaxEvents
	MaxStreamedHistoryBytes  int64 = api.MaxHistoryMaxBytes

	// DefaultMaxOrchestratorCompletionBytes leaves headroom below the modern DTS
	// orchestration-completion message limit.
	DefaultMaxOrchestratorCompletionBytes = 4_089_446
)

Variables

View Source 
var (
	// ErrTaskHubNotFound indicates that the configured task hub does not exist.
	ErrTaskHubNotFound = errors.New("task hub not found")
	// ErrStreamedHistoryLimitExceeded indicates that a worker refused to retain
	// an orchestration history beyond its configured safety limit.
	ErrStreamedHistoryLimitExceeded = errors.New("streamed orchestration history limit exceeded")
)
View Source 
var (
	ErrTaskHubGrpcWorkerAlreadyRunning = errors.New("gRPC worker is already running")
)

Functions

This section is empty.

Types

type TaskHubGrpcClient 

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

func NewTaskHubGrpcClient 

func NewTaskHubGrpcClient(cc grpc.ClientConnInterface, logger api.Logger, opts ...TaskHubGrpcClientOption) *TaskHubGrpcClient

NewTaskHubGrpcClient creates a client that can be used to manage orchestrations over a borrowed gRPC connection. The caller retains ownership of the connection; this client neither closes nor recreates it. DTS applications that want an owned, resilient management channel should use durabletaskscheduler.NewClient.

func (*TaskHubGrpcClient) CleanEntityStorage 

func (c *TaskHubGrpcClient) CleanEntityStorage(
	ctx context.Context,
	options ...api.CleanEntityStorageOptions,
) (*api.CleanEntityStorageResult, error)

CleanEntityStorage removes empty entities and releases orphaned locks.

func (*TaskHubGrpcClient) FetchOrchestrationMetadata 

func (c *TaskHubGrpcClient) FetchOrchestrationMetadata(ctx context.Context, id api.InstanceID, opts ...api.FetchOrchestrationMetadataOptions) (*api.OrchestrationMetadata, error)

FetchOrchestrationMetadata fetches metadata for the specified orchestration from the configured task hub.

api.ErrInstanceNotFound is returned when the specified orchestration doesn't exist.

func (*TaskHubGrpcClient) GetEntity 

func (c *TaskHubGrpcClient) GetEntity(
	ctx context.Context,
	entityID api.EntityID,
	options ...api.GetEntityOptions,
) (*api.EntityMetadata, error)

GetEntity retrieves metadata for an entity instance or returns nil when it doesn't exist. State is included unless ExcludeState is set.

func (*TaskHubGrpcClient) GetOrchestrationHistory 

func (c *TaskHubGrpcClient) GetOrchestrationHistory(
	ctx context.Context,
	id api.InstanceID,
	query api.HistoryQuery,
) (*api.OrchestrationHistory, error)

func (*TaskHubGrpcClient) ListInstanceIDs 

func (c *TaskHubGrpcClient) ListInstanceIDs(ctx context.Context, query api.InstanceIDQuery) (*api.InstanceIDQueryResult, error)

func (*TaskHubGrpcClient) PurgeInstances 

func (c *TaskHubGrpcClient) PurgeInstances(ctx context.Context, request api.PurgeInstancesRequest) (*api.PurgeInstancesResult, error)

func (*TaskHubGrpcClient) PurgeOrchestrationState 

func (c *TaskHubGrpcClient) PurgeOrchestrationState(ctx context.Context, id api.InstanceID, opts ...api.PurgeOptions) error

PurgeOrchestrationState deletes the state of the specified orchestration instance.

[api.api.ErrInstanceNotFound] is returned if the specified orchestration instance doesn't exist.

func (*TaskHubGrpcClient) QueryEntities 

func (c *TaskHubGrpcClient) QueryEntities(ctx context.Context, query api.EntityQuery) (*api.EntityQueryResults, error)

QueryEntities queries entities matching the supplied filters.

func (*TaskHubGrpcClient) QueryInstances 

func (c *TaskHubGrpcClient) QueryInstances(ctx context.Context, query api.OrchestrationQuery) (*api.OrchestrationQueryResult, error)

func (*TaskHubGrpcClient) RaiseEvent 

func (c *TaskHubGrpcClient) RaiseEvent(ctx context.Context, id api.InstanceID, eventName string, opts ...api.RaiseEventOptions) error

RaiseEvent sends an asynchronous event notification to a waiting orchestration.

func (*TaskHubGrpcClient) RestartInstance 

func (c *TaskHubGrpcClient) RestartInstance(ctx context.Context, id api.InstanceID, opts ...api.RestartOptions) (api.InstanceID, error)

func (*TaskHubGrpcClient) ResumeOrchestration 

func (c *TaskHubGrpcClient) ResumeOrchestration(ctx context.Context, id api.InstanceID, reason string) error

ResumeOrchestration resumes an orchestration instance that was previously suspended.

func (*TaskHubGrpcClient) RewindInstance 

func (c *TaskHubGrpcClient) RewindInstance(ctx context.Context, id api.InstanceID, opts ...api.RewindOptions) error

RewindInstance enqueues recovery of a failed orchestration. The service replaces its failed history with a new execution, preserving successful work and recursively rewinding failed sub-orchestrations.

A nil error means the request was enqueued, not that recovery completed. Observe a new execution and its terminal status separately. Canceling ctx cancels enqueueing; it does not cancel a rewind already accepted by the service.

func (*TaskHubGrpcClient) ScheduleNewOrchestration 

func (c *TaskHubGrpcClient) ScheduleNewOrchestration(ctx context.Context, orchestrator string, opts ...api.NewOrchestrationOptions) (api.InstanceID, error)

ScheduleNewOrchestration schedules a new orchestration instance with a specified set of options for execution.

func (*TaskHubGrpcClient) SignalEntity 

func (c *TaskHubGrpcClient) SignalEntity(ctx context.Context, entityID api.EntityID, operationName string, opts ...api.SignalEntityOptions) error

SignalEntity sends a fire-and-forget operation to an entity.

func (*TaskHubGrpcClient) StartWorkItemListener 

func (c *TaskHubGrpcClient) StartWorkItemListener(
	ctx context.Context,
	registry *task.TaskRegistry,
	workerOptions ...TaskHubGrpcWorkerOption,
) error

StartWorkItemListener preserves the original client API. New applications should create a dedicated TaskHubGrpcWorker, ideally with its own connection.

The listener borrows the client's connection, so it inherits the connection ownership limitations documented on NewTaskHubGrpcWorker.

func (*TaskHubGrpcClient) StopWorkItemListener 

func (c *TaskHubGrpcClient) StopWorkItemListener(ctx context.Context) error

StopWorkItemListener gracefully stops the compatibility listener.

func (*TaskHubGrpcClient) StreamOrchestrationHistory 

func (c *TaskHubGrpcClient) StreamOrchestrationHistory(
	ctx context.Context,
	id api.InstanceID,
	query api.HistoryQuery,
	handler api.HistoryEventHandler,
) error

func (*TaskHubGrpcClient) SuspendOrchestration 

func (c *TaskHubGrpcClient) SuspendOrchestration(ctx context.Context, id api.InstanceID, reason string) error

SuspendOrchestration suspends an orchestration instance, halting processing of its events until a "resume" operation resumes it.

Note that suspended orchestrations are still considered to be "running" even though they will not process events.

func (*TaskHubGrpcClient) TerminateOrchestration 

func (c *TaskHubGrpcClient) TerminateOrchestration(ctx context.Context, id api.InstanceID, opts ...api.TerminateOptions) error

TerminateOrchestration terminates a running orchestration by causing it to stop receiving new events and putting it directly into the TERMINATED state.

func (*TaskHubGrpcClient) WaitForOrchestrationCompletion 

func (c *TaskHubGrpcClient) WaitForOrchestrationCompletion(ctx context.Context, id api.InstanceID, opts ...api.FetchOrchestrationMetadataOptions) (*api.OrchestrationMetadata, error)

WaitForOrchestrationCompletion waits for an orchestration to complete and returns an api.OrchestrationMetadata object that contains metadata about the completed instance.

api.ErrInstanceNotFound is returned when the specified orchestration doesn't exist.

func (*TaskHubGrpcClient) WaitForOrchestrationStart 

func (c *TaskHubGrpcClient) WaitForOrchestrationStart(ctx context.Context, id api.InstanceID, opts ...api.FetchOrchestrationMetadataOptions) (*api.OrchestrationMetadata, error)

WaitForOrchestrationStart waits for an orchestration to start running and returns an api.OrchestrationMetadata object that contains metadata about the started instance.

api.ErrInstanceNotFound is returned when the specified orchestration doesn't exist.

type TaskHubGrpcClientOption 

type TaskHubGrpcClientOption func(*TaskHubGrpcClient)

func WithDataConverter 

func WithDataConverter(converter api.DataConverter) TaskHubGrpcClientOption

WithDataConverter configures application payload serialization.

func WithDefaultVersion 

func WithDefaultVersion(version string) TaskHubGrpcClientOption

WithDefaultVersion configures the version used when a top-level orchestration is scheduled without an explicit api.WithVersion option.

func WithLargePayloads 

func WithLargePayloads(options *api.LargePayloadOptions) TaskHubGrpcClientOption

WithLargePayloads configures externalization and hydration for management payloads.

type TaskHubGrpcWorker 

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

TaskHubGrpcWorker executes orchestration, activity, and entity work received from a DTS gRPC work-item stream (named TaskHubSidecarService in the wire contract).

func NewTaskHubGrpcWorker 

func NewTaskHubGrpcWorker(
	cc grpc.ClientConnInterface,
	registry *task.TaskRegistry,
	logger api.Logger,
	opts ...TaskHubGrpcWorkerOption,
) (*TaskHubGrpcWorker, error)

NewTaskHubGrpcWorker creates a worker that borrows a caller-owned connection.

The worker never closes or replaces the supplied connection: a reconnect only recreates the work-item stream on that same connection. Recovery is therefore limited to what the caller's connection can do on its own. A connection that is permanently wedged (for example one whose credentials expired, or one whose endpoint moved) keeps producing poisoned streams and the worker keeps retrying on the escalating reconnect schedule without ever obtaining a fresh channel. Use NewTaskHubGrpcWorkerWithConnectionFactory, or the Durable Task Scheduler NewWorker, when the worker should own and recreate its channels.

func NewTaskHubGrpcWorkerWithConnectionFactory 

func NewTaskHubGrpcWorkerWithConnectionFactory(
	factory TaskHubGrpcWorkerConnectionFactory,
	registry *task.TaskRegistry,
	logger api.Logger,
	opts ...TaskHubGrpcWorkerOption,
) (*TaskHubGrpcWorker, error)

NewTaskHubGrpcWorkerWithConnectionFactory creates a worker whose connection can be recreated after disconnects. Each returned closer is called only after all work dispatched through that connection has completed or been abandoned.

func (*TaskHubGrpcWorker) Run 

func (w *TaskHubGrpcWorker) Run(ctx context.Context) error

Run connects to the service, performs the Hello handshake, and blocks until the worker stops.

func (*TaskHubGrpcWorker) Running 

func (w *TaskHubGrpcWorker) Running() bool

func (*TaskHubGrpcWorker) Shutdown 

func (w *TaskHubGrpcWorker) Shutdown(ctx context.Context) error

Shutdown stops intake and waits for in-flight work to finish. If ctx expires, in-flight execution and completion RPCs are canceled.

func (*TaskHubGrpcWorker) Start 

func (w *TaskHubGrpcWorker) Start(ctx context.Context) error

Start connects to the service, performs the Hello handshake, and starts the worker in the background.

func (*TaskHubGrpcWorker) Wait 

func (w *TaskHubGrpcWorker) Wait(ctx context.Context) error

Wait waits for a worker started with Start to stop.

type TaskHubGrpcWorkerConnectionFactory 

type TaskHubGrpcWorkerConnectionFactory func(context.Context) (grpc.ClientConnInterface, io.Closer, error)

TaskHubGrpcWorkerConnectionFactory creates a gRPC connection for a worker stream generation. A non-nil closer transfers ownership to the worker.

type TaskHubGrpcWorkerOption 

type TaskHubGrpcWorkerOption func(*taskHubGrpcWorkerOptions) error

func CombineTaskHubGrpcWorkerOptions 

func CombineTaskHubGrpcWorkerOptions(options ...TaskHubGrpcWorkerOption) TaskHubGrpcWorkerOption

CombineTaskHubGrpcWorkerOptions combines worker options into one option.

func WithAutoWorkItemFilters 

func WithAutoWorkItemFilters() TaskHubGrpcWorkerOption

WithAutoWorkItemFilters derives task-name and task-version filters from the registry. Explicit WithWorkItemFilters configuration takes precedence.

func WithMaxConcurrentActivityWorkItems 

func WithMaxConcurrentActivityWorkItems(n int) TaskHubGrpcWorkerOption

func WithMaxConcurrentEntityWorkItems 

func WithMaxConcurrentEntityWorkItems(n int) TaskHubGrpcWorkerOption

WithMaxConcurrentEntityWorkItems limits concurrent entity batch execution.

func WithMaxConcurrentOrchestrationWorkItems 

func WithMaxConcurrentOrchestrationWorkItems(n int) TaskHubGrpcWorkerOption

func WithMaxOrchestratorCompletionBytes 

func WithMaxOrchestratorCompletionBytes(n int) TaskHubGrpcWorkerOption

WithMaxOrchestratorCompletionBytes lowers the post-externalization response bound used before completing an orchestration work item. Multiple options are monotonic: a later option cannot raise an earlier transport-derived cap.

func WithMaxStreamedHistoryBytes 

func WithMaxStreamedHistoryBytes(n int64) TaskHubGrpcWorkerOption

WithMaxStreamedHistoryBytes limits the protobuf bytes retained while receiving a streamed orchestration history.

func WithMaxStreamedHistoryEvents 

func WithMaxStreamedHistoryEvents(n int) TaskHubGrpcWorkerOption

WithMaxStreamedHistoryEvents limits the number of history events retained while receiving a streamed orchestration history.

func WithMaximumTimerInterval 

func WithMaximumTimerInterval(interval time.Duration) TaskHubGrpcWorkerOption

WithMaximumTimerInterval configures deterministic splitting of long durable timers. Changing this value is replay-breaking for affected in-flight orchestrations.

func WithScheduledTaskCapability 

func WithScheduledTaskCapability(enabled bool) TaskHubGrpcWorkerOption

WithScheduledTaskCapability controls scheduled-task capability advertisement.

func WithTaskExecutorOptions 

func WithTaskExecutorOptions(options ...task.TaskExecutorOption) TaskHubGrpcWorkerOption

WithTaskExecutorOptions configures the task executor used by the gRPC worker. This allows task options such as task.WithVersioning to participate in DTS dispatch.

func WithTaskVersioning 

func WithTaskVersioning(versioning task.VersioningOptions) TaskHubGrpcWorkerOption

WithTaskVersioning configures worker version acceptance and the default version used for sub-orchestrations.

func WithUnversionedActivityNames 

func WithUnversionedActivityNames(names ...string) TaskHubGrpcWorkerOption

WithUnversionedActivityNames allows explicitly unversioned system activities to remain routable when strict worker versioning is enabled. System orchestrations run unversioned and an activity inherits its caller's version, so a system component's activities must be advertised unversioned too.

func WithUnversionedOrchestratorNames 

func WithUnversionedOrchestratorNames(names ...string) TaskHubGrpcWorkerOption

WithUnversionedOrchestratorNames allows explicitly unversioned system orchestrators to remain routable when strict worker versioning is enabled.

func WithWorkItemFilters 

func WithWorkItemFilters(filters *WorkItemFilters) TaskHubGrpcWorkerOption

WithWorkItemFilters restricts orchestration/activity names and versions and entity names accepted by the worker. A nil per-kind list means no restriction for that kind. RejectAll* explicitly rejects a kind.

func WithWorkerCapabilities 

func WithWorkerCapabilities(capabilities ...WorkerCapability) TaskHubGrpcWorkerOption

WithWorkerCapabilities explicitly configures the capabilities advertised to DTS.

func WithWorkerDataConverter 

func WithWorkerDataConverter(converter api.DataConverter) TaskHubGrpcWorkerOption

WithWorkerDataConverter configures application payload serialization.

func WithWorkerHelloTimeout 

func WithWorkerHelloTimeout(timeout time.Duration) TaskHubGrpcWorkerOption

func WithWorkerLargePayloads 

func WithWorkerLargePayloads(options *api.LargePayloadOptions) TaskHubGrpcWorkerOption

WithWorkerLargePayloads configures worker payload hydration/externalization and advertises support.

func WithWorkerRPCTimeout 

func WithWorkerRPCTimeout(timeout time.Duration) TaskHubGrpcWorkerOption

func WithWorkerReconnectBackoff 

func WithWorkerReconnectBackoff(baseDelay, maxDelay time.Duration) TaskHubGrpcWorkerOption

WithWorkerReconnectBackoff configures the reconnect schedule. Nominal delays double from baseDelay, receive ±25% jitter, and remain within [baseDelay, maxDelay].

func WithWorkerSilentDisconnectTimeout 

func WithWorkerSilentDisconnectTimeout(timeout time.Duration) TaskHubGrpcWorkerOption

func WithWorkerTransientRetryPolicy 

func WithWorkerTransientRetryPolicy(maxAttempts int, baseDelay, maxDelay time.Duration) TaskHubGrpcWorkerOption

WithWorkerTransientRetryPolicy configures the deterministic retry schedule for completion and abandon RPCs. Delays double from baseDelay and are always within [baseDelay, maxDelay].

type WorkItemFilter 

type WorkItemFilter struct {
	Name     string
	Versions []string
}

type WorkItemFilters 

type WorkItemFilters struct {
	Orchestrations          []WorkItemFilter
	Activities              []WorkItemFilter
	Entities                []string
	RejectAllOrchestrations bool
	RejectAllActivities     bool
	RejectAllEntities       bool
}

type WorkerCapability 

type WorkerCapability = protos.WorkerCapability

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.