dex

import "github.com/superdurable/dex/sdk-go/dex"

Index

Constants

DefaultGrpcMaxMessageBytes is the gRPC receive limit that a Client or Worker uses when ClientOptions.GrpcMaxMessageBytes or WorkerOptions.GrpcMaxMessageBytes is zero.

The receive limit applies to every message the SDK receives: Client responses such as InvokeRPC results and state reads, responses to the Worker's own FlowService calls such as LoadBlobs, and the Step and RPC invocations that the Dex server sends to the Worker. The default is 32 MiB, which matches the Dex server's default api.grpcMaxMessageBytes. With the default, the SDK keeps gRPC's unlimited send limit, so the Dex server's receive limit bounds what the SDK sends. A positive GrpcMaxMessageBytes limits both directions.

A message carries one or more application values plus an envelope of metadata. Keep the limit above the largest application value by an explicit margin; the default leaves 16 MiB of headroom for values up to 16 MiB. When an application needs larger values, raise both GrpcMaxMessageBytes and the server's api.grpcMaxMessageBytes, because each hop applies the smaller limit of its ends.

client, err := dex.NewClient(registry, cache, dex.ClientOptions{
	GrpcMaxMessageBytes: 2 * dex.DefaultGrpcMaxMessageBytes,
})

const DefaultGrpcMaxMessageBytes = 32 << 20

WorkQueuePermissionsIndexKey is the Temporal KeywordList Search Attribute containing permissions for Actions that have been available in the current Flow execution.

The index is a Work Queue discovery projection, not an authorization result. An Action gateway must still authorize the selected RPC against its required permission.

const WorkQueuePermissionsIndexKey = "DexWorkQueuePermissions"

func ErrorWithStack

func ErrorWithStack(err error) error

ErrorWithStack wraps err with a stack captured at this call site.

Use it when returning a plain Go error from waitFor or execute and you want Dex to record an origin stack. Plain errors are not Dex-owned, so capture is opt-in; without this wrapper Dex reports that no application stack was captured instead of inventing a Worker wrap-site stack.

Prefer nesting ErrorWithStack inside RetryAfter so RetryAfter remains the protocol marker and the stack lives on the cause:

return nil, dex.RetryAfter(30*time.Second, dex.ErrorWithStack(err))

Unwrap returns the original err. Error returns the original message. Formatting with %+v includes the captured stack. A nil err returns nil.

func GetFinalFlowType

func GetFinalFlowType(flow Flow) string

GetFinalFlowType returns the Flow type that NewRegistry registers for a Flow.

A non-empty GetFlowType override wins. Otherwise the Flow type is the Go type name without its package or pointer, so OrderFlow{} and &OrderFlow{} from package orders both return "OrderFlow". Use it in tests and tools that need the registered name, such as a SearchFlows query on FlowType.

query := "FlowType = '" + dex.GetFinalFlowType(&orders.OrderFlow{}) + "'"

A generic or unnamed Go type has no default Flow type. Such a Flow must implement GetFlowType, and NewRegistry rejects it otherwise.

flow is the Flow value to name. It is not retained.

Returns the override when it is non-empty, otherwise the Go type name without its package or pointer. Returns an empty string when there is no override and the Go type is generic or unnamed.

func GetFinalStepType

func GetFinalStepType[IN any](step Step[IN]) string

GetFinalStepType returns the Step type that NewRegistry registers for a Step.

A non-empty GetStepType override wins. Otherwise the Step type is the Go type name without its package or pointer, so shipOrder{} and &shipOrder{} both return "shipOrder". Use it in tests and tools that need the registered name, such as a StepExecutionID passed to SkipTimer or WaitForStepCompletion.

stepType := dex.GetFinalStepType[ShipInput](shipOrder{})
executionID := dex.StepExecutionID{StepType: stepType}

A generic or unnamed Go type has no default Step type. Such a Step must implement GetStepType, and NewRegistry rejects it otherwise.

IN is the Step input type. step is the Step value to name. It is not retained.

Returns the override when it is non-empty, otherwise the Go type name without its package or pointer. Returns an empty string when there is no override and the Go type is generic or unnamed.

func SubFlowID

func SubFlowID(ctx Context, index ...int) (string, error)

SubFlowID returns one server-generated SubFlow Flow ID from the current Execute invocation.

Omit index for the first SubFlow. The ID remains addressable after AnyOf completes and can be passed to Client.StopFlow to stop a still-running loser.

type APIKeyCredentialOption

APIKeyCredentialOption configures an APIKeyCredential.

Create options with WithAPIKeyExchangeTimeout and WithAPIKeyHTTPClient, and pass them to APIKeyCredential. Options apply in order, so a later option of the same kind replaces an earlier one. See APIKeyCredential for a complete example.

type APIKeyCredentialOption interface {
    // contains filtered or unexported methods
}

func WithAPIKeyExchangeTimeout

func WithAPIKeyExchangeTimeout(timeout time.Duration) APIKeyCredentialOption

WithAPIKeyExchangeTimeout bounds each token exchange of an APIKeyCredential.

The timeout covers connecting to the token endpoint, sending the request, and reading the complete response. The default is 10 seconds. A Dex call that has no valid cached token waits at most this long for the exchange, and less when its own context ends first. The timeout applies through the request context, so an HTTP client supplied with WithAPIKeyHTTPClient keeps its own Timeout as an additional limit.

credential, err := dex.APIKeyCredential(tokenURL, serverID, apiKey,
	dex.WithAPIKeyExchangeTimeout(3*time.Second))

timeout must be positive. APIKeyCredential returns an error for a zero or negative timeout.

func WithAPIKeyHTTPClient

func WithAPIKeyHTTPClient(httpClient *http.Client) APIKeyCredentialOption

WithAPIKeyHTTPClient sets the HTTP client that an APIKeyCredential uses for token exchanges.

Use it to route exchanges through a proxy, to trust a private certificate authority, or to reach an in-process token endpoint in tests. By default the credential uses a client built on http.DefaultTransport, which honors the standard proxy environment variables. The credential copies the client and replaces its CheckRedirect so that it never follows a redirect. A redirect response fails the exchange instead, so the API key is sent only to the configured token URL. The credential never closes the client or its transport; the caller keeps ownership of both.

httpClient := &http.Client{Transport: &http.Transport{Proxy: http.ProxyURL(proxyURL)}}
credential, err := dex.APIKeyCredential(tokenURL, serverID, apiKey,
	dex.WithAPIKeyHTTPClient(httpClient))

httpClient must not be nil. APIKeyCredential returns an error for a nil client.

type ActionConditionDef

ActionConditionDef is the schema-erased availability condition for an Action.

Applications create conditions with WhenAttributeMatches. This interface is sealed so only SDK conditions can be registered.

type ActionConditionDef interface {
    // contains filtered or unexported methods
}

func WhenAttributeMatches

func WhenAttributeMatches[T any](attribute Attribute[T], matches ...AttributeMatch[T]) ActionConditionDef

WhenAttributeMatches creates an Action condition for one scalar Attribute.

At least one match is required. Every match must be created with AttributeMatchEqual. Multiple matches use OR semantics. A missing or deleted Attribute does not match. Supported Attribute types are strings, booleans, integers, unsigned integers that fit int64, and finite floating-point values.

The Attribute must be registered in the same Flow as the Action RPC.

type ActionDef

ActionDef is the schema-erased registration definition for an Action RPC.

Applications create an Action with DefineAction and assign it to [RPCOptions.Action]. This interface is sealed so only SDK definitions can be registered.

type ActionDef interface {
    // contains filtered or unexported methods
}

func DefineAction

func DefineAction(label string, condition ActionConditionDef, options ...ActionOption) ActionDef

DefineAction creates the immutable contract for one Action RPC.

label is the user-facing Action label. condition determines whether the Action is available from the post-invocation Attribute state. Exactly one ActionRequiresPermission option is required.

Example:

action := dex.DefineAction(
	"Approve",
	dex.WhenAttributeMatches(
		caseStatus,
		dex.AttributeMatchEqual("awaiting-manager"),
	),
	dex.ActionRequiresPermission("refund.approve"),
)

type ActionOption

ActionOption configures an Action definition.

Use ActionRequiresPermission to declare the permission required by the Action. Custom implementations are not supported.

type ActionOption interface {
    // contains filtered or unexported methods
}

func ActionRequiresPermission

func ActionRequiresPermission(permission string) ActionOption

ActionRequiresPermission declares the single permission required by an Action.

The permission uses lowercase letter and number segments separated by dots or hyphens, such as "refund.approve" or "permission-x". A Flow may register multiple Actions with the same permission; the Work Queue projection removes duplicates.

type ActiveStepSearchMode

ActiveStepSearchMode controls which active Step types Dex indexes for Flow search.

type ActiveStepSearchMode uint8


const (
    // SearchActiveStepsDefault uses the server default indexing policy.
    SearchActiveStepsDefault ActiveStepSearchMode = iota
    // SearchAllActiveSteps indexes every active Step, including execute-only Steps.
    SearchAllActiveSteps
    // SearchActiveStepsWithWaitFor indexes a Step only after WaitFor runs.
    SearchActiveStepsWithWaitFor
    // DisableActiveStepSearch disables active-Step indexing for the Flow.
    DisableActiveStepSearch
)

type AlreadyStartedOptions

AlreadyStartedOptions configures StartFlow behavior when the Flow ID already exists.

type AlreadyStartedOptions struct {
    // IgnoreError returns the existing run ID instead of FlowAlreadyStartedError.
    // It applies only when StartFlowOptions.RequestID matches the request ID that started the
    // existing run. A different or SDK-generated request ID still returns FlowAlreadyStartedError.
    IgnoreError bool
}

type Attribute

Attribute defines a typed persisted value.

Declare an Attribute once, include it in the Flow's PersistenceSchema, and reuse the same value from Step and RPC code. Values use the SDK's primitive and JSON encoding rules. String values require valid UTF-8; use []byte for arbitrary bytes.

Example:

var orderStatus = dex.DefineAttribute[string](
	"order-status",
	dex.Indexed(dex.AttributeIndex{Type: dex.IndexKeyword}),
)

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

func DefineAttribute

func DefineAttribute[T any](key string, options ...AttributeOption) Attribute[T]

DefineAttribute creates a typed Attribute with a stable name and definition options. "/" is reserved as the AttributeMap separator and is prohibited in Attribute names. The definition does not perform I/O; register it through PersistenceSchema before use.

func (Attribute[T]) AttributeName

func (a Attribute[T]) AttributeName() string

AttributeName returns the stable name sent to Dex and used in PersistenceSchema.

func (Attribute[T]) Delete

func (a Attribute[T]) Delete(ctx Context) error

Delete stages removal of this Attribute. Deleting a missing value is valid. It returns an error when ctx is not an SDK invocation.

func (Attribute[T]) Get

func (a Attribute[T]) Get(ctx Context) (value T, err error)

Get returns the current value. Missing values return AttributeNotFoundError. It returns an error when ctx is not an SDK invocation or the value cannot be decoded.

func (Attribute[T]) Set

func (a Attribute[T]) Set(ctx Context, value T) error

Set stages value for durable persistence when the current Step or RPC invocation succeeds. It returns an error when ctx is not an SDK invocation or value cannot be encoded or indexed.

type AttributeDef

AttributeDef is the interface of Attribute, without Go's generic So that internal sdk can use it to workaround Go's generic limitations

AttributeDef is internal to the SDK. Applications create values with DefineAttribute or DefineAttributeMap, then pass them to PersistenceSchema and Client methods.

type AttributeDef interface {
    // contains filtered or unexported methods
}

type AttributeIndex

AttributeIndex configures visibility indexing; datetime values use time.Time or RFC3339Nano strings. An empty IndexKey uses the Attribute name for single values and a derived physical name for map instances.

type AttributeIndex struct {
    // Type selects the required value representation.
    Type IndexType
    // IndexKey overrides the physical search field; empty uses the Attribute-derived key.
    IndexKey string
}

type AttributeLock

AttributeLock identifies one Attribute or Attribute-map instance lock. Create locks with LockAttribute or LockAttributeMap, then place them in StepOptions or RPCOptions.

type AttributeLock interface {
    // contains filtered or unexported methods
}

func LockAttribute

func LockAttribute[T any](attribute Attribute[T]) AttributeLock

LockAttribute creates a lock for the single value represented by attribute.

func LockAttributeMap

func LockAttributeMap[T any](attribute AttributeMap[T], instance string) AttributeLock

LockAttributeMap creates a lock scoped to one Attribute-map instance.

type AttributeMap

AttributeMap defines keyed typed persisted values.

Register the map once in PersistenceSchema, then supply an instance string for every access and lock. Slash is prohibited in instance keys because it is a reserved character. String values require valid UTF-8; use []byte for arbitrary bytes.

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

func DefineAttributeMap

func DefineAttributeMap[T any](name string, options ...AttributeOption) AttributeMap[T]

DefineAttributeMap creates a typed Attribute map with a stable name and definition options. "/" is reserved as the AttributeMap separator and is prohibited in AttributeMap names. The definition performs no I/O; register it through PersistenceSchema before use.

func (AttributeMap[T]) AllInstanceKeys

func (a AttributeMap[T]) AllInstanceKeys(ctx Context) []string

AllInstanceKeys returns existing instance keys in ascending order, including buffered writes.

func (AttributeMap[T]) AttributeName

func (a AttributeMap[T]) AttributeName() string

AttributeName returns the stable shared name of this Attribute map.

func (AttributeMap[T]) Delete

func (a AttributeMap[T]) Delete(ctx Context, instance string) error

Delete stages removal of one map instance. Deleting a missing instance is valid. It returns an error when ctx is not an SDK invocation.

func (AttributeMap[T]) Get

func (a AttributeMap[T]) Get(ctx Context, instance string) (value T, err error)

Get returns the current map value. Missing instances return AttributeNotFoundError. It returns an error when ctx is invalid or the value cannot be decoded.

func (AttributeMap[T]) Load

func (a AttributeMap[T]) Load(instance string) AttributeMapLoad

Load selects one slash-free logical instance for a handler snapshot. Loading does not lock the instance or prevent concurrent changes.

func (AttributeMap[T]) MapSize

func (a AttributeMap[T]) MapSize(ctx Context) int

MapSize returns the number of existing instances, including writes buffered by this invocation.

func (AttributeMap[T]) Set

func (a AttributeMap[T]) Set(ctx Context, instance string, value T) error

Set stages value for one map instance when the invocation succeeds. It returns an error for an invalid context or a value that cannot be encoded or indexed.

type AttributeMapLoad

AttributeMapLoad selects one AttributeMap instance for a handler invocation. Create values with AttributeMap.Load and place them in the matching invocation options.

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

type AttributeMapNotLoadedError

AttributeMapNotLoadedError reports an AttributeMap entry read omitted by invocation options.

type AttributeMapNotLoadedError struct {
    // Name is the AttributeMap name or its physical instance name.
    Name string
}

func (*AttributeMapNotLoadedError) Error

func (e *AttributeMapNotLoadedError) Error() string

Error describes the missing AttributeMap load.

type AttributeMatch

AttributeMatch describes a scalar Attribute predicate.

Create a match with an AttributeMatch factory and pass it to Client.WaitForAttributeMatch or Client.WaitForAttributeMapInstanceMatch. The operand type must match the registered Attribute type. String and boolean Attributes support equality operators. Integer and floating-point Attributes support every operator.

Example:

match := dex.AttributeMatchGreaterThan(int64(3))

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

func AttributeMatchEqual

func AttributeMatchEqual[T any](operand T) AttributeMatch[T]

AttributeMatchEqual creates a match that succeeds when the Attribute equals operand.

String, boolean, integer, and floating-point operands are supported.

func AttributeMatchGreaterThan

func AttributeMatchGreaterThan[T any](operand T) AttributeMatch[T]

AttributeMatchGreaterThan creates a match that succeeds when the Attribute is greater than operand.

The operand must be an integer or finite floating-point value.

func AttributeMatchGreaterThanOrEqual

func AttributeMatchGreaterThanOrEqual[T any](operand T) AttributeMatch[T]

AttributeMatchGreaterThanOrEqual creates a greater-than-or-equal Attribute match.

The operand must be an integer or finite floating-point value.

func AttributeMatchLessThan

func AttributeMatchLessThan[T any](operand T) AttributeMatch[T]

AttributeMatchLessThan creates a match that succeeds when the Attribute is less than operand.

The operand must be an integer or finite floating-point value.

func AttributeMatchLessThanOrEqual

func AttributeMatchLessThanOrEqual[T any](operand T) AttributeMatch[T]

AttributeMatchLessThanOrEqual creates a less-than-or-equal Attribute match.

The operand must be an integer or finite floating-point value.

func AttributeMatchNotEqual

func AttributeMatchNotEqual[T any](operand T) AttributeMatch[T]

AttributeMatchNotEqual creates a match that succeeds when the Attribute differs from operand.

A missing Attribute does not satisfy this match. String, boolean, integer, and floating-point operands are supported.

type AttributeMatchDef

AttributeMatchDef is the schema-erased form of an AttributeMatch.

Applications create matches with the AttributeMatch factory functions. This interface is sealed so only SDK-defined match values can implement it.

type AttributeMatchDef interface {
    // contains filtered or unexported methods
}

type AttributeNotFoundError

AttributeNotFoundError reports a missing invocation attribute.

type AttributeNotFoundError struct {
    // AttributeName is the stable Attribute or AttributeMap name.
    AttributeName string
    // Instance is the missing map instance, or empty for a single Attribute.
    Instance string
}

func (*AttributeNotFoundError) Error

func (e *AttributeNotFoundError) Error() string

Error identifies the missing Attribute or Attribute-map instance.

type AttributeOption

AttributeOption configures an Attribute or AttributeMap definition. Use Indexed, SyncToAttributeStore, and HardDeleteOnNull to create values; custom implementations are not supported.

type AttributeOption interface {
    // contains filtered or unexported methods
}

func HardDeleteOnNull

func HardDeleteOnNull() AttributeOption

HardDeleteOnNull keeps an Attribute's values out of Flow history and permanently deletes them on Delete.

Use it for short-lived secrets, such as an API key that a Flow uses a few times and then discards. Dex stores every value of the Attribute only in its blob store, whatever the value size, and Flow history keeps only a reference. Delete, or Set with a nil value, permanently deletes the stored value, so neither Dex APIs nor Flow history can recover it afterward. Step and RPC handlers receive the value inline, and the SDK never writes it to the local blob cache.

Only string and object values are supported, including []byte. Registration rejects integer, unsigned integer, floating-point, and Boolean value types. An interface value type is checked when a value is encoded. The option cannot be combined with Indexed or SyncToAttributeStore, and the Attribute cannot be used by WaitForAttributeMatch, WaitForAttributeMapInstanceMatch, or an Action condition. The Dex Server must enable its blob store and support Server protocol version 3 or later. Otherwise StartFlow returns an error, and a Step or RPC that writes the Attribute fails.

Each Flow keeps one stored value per Attribute key and overwrites it in place. This has user-visible consequences:

A value remains stored until it is deleted, or until Dex removes the Flow after its retention period. Blob store backups or bucket versioning outside Dex can keep copies.

Example:

var ModelAPIKey = dex.DefineAttribute[string]("model-api-key", dex.HardDeleteOnNull())

type CallModelStep struct {
	dex.StepDefaultsNoWaitFor[CallModelInput]
}

func (CallModelStep) Execute(ctx dex.Context, input CallModelInput) (*dex.StepDecision, error) {
	apiKey, err := ModelAPIKey.Get(ctx)
	if err != nil {
		return nil, err
	}
	if err := callModel(ctx, apiKey, input.Prompt); err != nil {
		return nil, err
	}
	if err := ModelAPIKey.Delete(ctx); err != nil {
		return nil, err
	}
	return dex.GracefulComplete(nil), nil
}

Pass the secret when the Flow starts with InitialAttribute(ModelAPIKey, apiKey) in StartFlowOptions.Attributes, and include ModelAPIKey in the Flow's PersistenceSchema.

func Indexed

func Indexed(index AttributeIndex) AttributeOption

Indexed enables search indexing with index on an Attribute or AttributeMap definition.

func SyncToAttributeStore

func SyncToAttributeStore() AttributeOption

SyncToAttributeStore projects every write through the Flow's configured Attribute Store. Projection is asynchronous and latest-state only. Deletes write SQL NULL, and failures do not roll back Flow Attributes.

type AttributeRead

AttributeRead selects one registered Attribute or AttributeMap instance for Client.GetAttributes.

Create an AttributeRead with ReadAttribute or ReadAttributeMapInstance. Each AttributeRead pairs a definition with a destination for its typed value. GetAttributes writes the destination only when the value is set and the whole call succeeds. An unset value leaves the destination unchanged. Applications cannot implement AttributeRead. See Client.GetAttributes for an example.

type AttributeRead interface {
    // contains filtered or unexported methods
}

func ReadAttribute

func ReadAttribute[T any](attribute Attribute[T], destination *T) AttributeRead

ReadAttribute selects a registered Attribute and the destination for its value.

Pass the result to Client.GetAttributes. When the Attribute is set and the call succeeds, GetAttributes stores the decoded value in destination. When the Attribute is not set, destination keeps its previous value. Use AttributeReadResult.IsSet to tell an unset Attribute from a stored zero value. See Client.GetAttributes for an example.

Type parameter T is the Attribute value type. It must match the T of the Attribute definition registered in the Client's Registry.

Parameters:

Returns an AttributeRead for Client.GetAttributes. ReadAttribute does not validate its inputs or contact the server. GetAttributes reports invalid inputs.

func ReadAttributeMapInstance

func ReadAttributeMapInstance[T any](attributeMap AttributeMap[T], instance string, destination *T) AttributeRead

ReadAttributeMapInstance selects one registered AttributeMap instance and the destination for its value.

Pass the result to Client.GetAttributes. When the instance is set and the call succeeds, GetAttributes stores the decoded value in destination. When the instance is not set, destination keeps its previous value. Use AttributeReadResult.IsMapInstanceSet to tell an unset instance from a stored zero value. See Client.GetAttributes for an example.

Type parameter T is the AttributeMap value type. It must match the T of the AttributeMap definition registered in the Client's Registry.

Parameters:

Returns an AttributeRead for Client.GetAttributes. ReadAttributeMapInstance does not validate its inputs or contact the server. GetAttributes reports invalid inputs.

type AttributeReadResult

AttributeReadResult reports which reads of one Client.GetAttributes call found a set value.

GetAttributes returns an AttributeReadResult with every successful call. Most callers ignore it and rely on the destinations. Use it when an unset value must be distinguished from a stored zero value. The zero value is valid and reports every Attribute and AttributeMap instance as not set. GetAttributes returns the zero value with every error. See Client.GetAttributes for an example.

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

func (AttributeReadResult) IsMapInstanceSet

func (result AttributeReadResult) IsMapInstanceSet(attributeMap AttributeDef, instance string) bool

IsMapInstanceSet reports whether the call read one AttributeMap instance and found a set value.

It returns false when the instance was not set. It also returns false when the instance was not part of the call, attributeMap is nil, or attributeMap is not an AttributeMap. attributeMap must equal the definition passed to ReadAttributeMapInstance. Another definition with the same name, such as one with a different value type, reports false.

Parameters:

Returns true only when GetAttributes stored this instance's value in its destination.

func (AttributeReadResult) IsSet

func (result AttributeReadResult) IsSet(attribute AttributeDef) bool

IsSet reports whether the call read attribute and found a set value.

It returns false when attribute was not set. It also returns false when attribute was not part of the call, is nil, or is an AttributeMap. attribute must equal the definition passed to ReadAttribute. Another definition with the same name, such as one with a different value type, reports false. Use IsMapInstanceSet for an AttributeMap instance.

Parameters:

Returns true only when GetAttributes stored this Attribute's value in its destination.

type BlobMessageTooLargeError

BlobMessageTooLargeError reports a blob value that alone exceeds a gRPC message size limit.

Dex stores large Attribute, Channel, and RPC values as blobs and the SDK hydrates them with LoadBlobs. The SDK first requests every uncached blob in one LoadBlobs call. When that response exceeds a gRPC message size limit, the SDK splits the request in half and retries each half, so the batch size never needs tuning. This error means a single blob still does not fit in one response.

The limit that rejected the response is either the SDK's GrpcMaxMessageBytes on ClientOptions or WorkerOptions, or the Dex server's api.grpcMaxMessageBytes. Err keeps gRPC's RESOURCE_EXHAUSTED status, whose message states the response size and the limit. Raise the smaller limit above the value size plus envelope headroom, or split the value across several Attributes or AttributeMap instances.

A Client call returns this error wrapped in a ServiceError. A Worker reports it as an INTERNAL Step or RPC failure whose detail contains this message.

var tooLarge *dex.BlobMessageTooLargeError
if errors.As(err, &tooLarge) {
	log.Printf("Flow %s blob %s exceeds the gRPC limit: %v",
		tooLarge.FlowID, tooLarge.BlobID, tooLarge.Err)
}

type BlobMessageTooLargeError struct {
    // FlowID is the Flow ID that owns the blob.
    FlowID string
    // BlobID is the Dex blob ID of the value that did not fit.
    BlobID string
    // Err is the gRPC RESOURCE_EXHAUSTED status returned for the single-blob LoadBlobs call.
    Err error
}

func (*BlobMessageTooLargeError) Error

func (e *BlobMessageTooLargeError) Error() string

Error identifies the Flow ID, blob ID, and gRPC size failure.

func (*BlobMessageTooLargeError) Unwrap

func (e *BlobMessageTooLargeError) Unwrap() error

Unwrap returns the gRPC RESOURCE_EXHAUSTED status.

type BufferedTextStream

BufferedTextStream batches text chunks before appending them to a Step Stream.

Create one writer per logical text feed with NewBufferedTextStream. Write preserves every byte and appends the current batch when its one-second default interval or 16 KiB default threshold is reached. The invocation flushes and closes the writer before its final result or error is sent. Stream Store failures remain unacknowledged, matching Stream.Write.

progress, err := dex.NewBufferedTextStream(ctx, Thinking)
if err != nil {
	return nil, err
}
if err := progress.Write(delta); err != nil {
	return nil, err
}

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

func NewBufferedTextStream

func NewBufferedTextStream(ctx Context, stream Stream[string], options ...BufferedTextStreamOption) (*BufferedTextStream, error)

NewBufferedTextStream creates an invocation-managed writer for a registered string Stream.

The default flush interval is one second and the default soft size threshold is 16 KiB of UTF-8 text. The Context must belong to an active WaitFor or Execute invocation. The SDK automatically flushes remaining text and stops the timer before sending that invocation's final result or error.

The returned writer is safe for concurrent Write calls. Reuse it for the entire logical feed; creating multiple writers gives each writer an independent buffer and timer.

Returns an error for an inactive, RPC, or Flow-timeout Context, an unregistered Stream, or invalid options.

func (*BufferedTextStream) Write

func (writer *BufferedTextStream) Write(chunk string) error

Write appends one text chunk to the current batch.

Empty chunks are ignored. Write flushes immediately when the soft size threshold is reached. It otherwise returns after buffering locally; it does not wait for the interval or Stream Store. A timer or transport failure is returned by the next Write or invocation finalization.

type BufferedTextStreamOption

BufferedTextStreamOption configures a BufferedTextStream.

type BufferedTextStreamOption interface {
    // contains filtered or unexported methods
}

func BufferedTextStreamFlushInterval

func BufferedTextStreamFlushInterval(value time.Duration) BufferedTextStreamOption

BufferedTextStreamFlushInterval overrides the one-second batch interval.

The duration must be positive. A non-empty buffer is appended when the interval elapses even if the handler does not produce another chunk.

func BufferedTextStreamMaxBytes

func BufferedTextStreamMaxBytes(value int) BufferedTextStreamOption

BufferedTextStreamMaxBytes overrides the 16 KiB soft UTF-8 size threshold.

The value must be positive. A chunk is never split, so an emitted batch may exceed this threshold by the size of its final chunk.

type Channel

Channel defines one durable FIFO queue of typed messages.

Add the Channel to PersistenceSchema. Clients and handlers may publish messages; Step WaitFor methods create Conditions that wait on queue-size bounds and later decode consumed results.

Example:

var approvals = dex.DefineChannel[Approval]("approvals")
return dex.AnyOf(approvals.ForOne(), dex.Timer(5*time.Minute)), nil

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

func DefineChannel

func DefineChannel[T any](name string) Channel[T]

DefineChannel creates a typed Channel with a stable name without performing I/O. "/" is reserved as the ChannelMap separator and is prohibited in Channel names.

func (Channel[T]) AtLeast

func (c Channel[T]) AtLeast(count int, options ...ConditionOption) Condition

AtLeast returns a Condition satisfied when at least count messages are queued.

func (Channel[T]) AtLeastAtMost

func (c Channel[T]) AtLeastAtMost(atLeast int, atMost int, options ...ConditionOption) Condition

AtLeastAtMost waits for atLeast messages, then consumes up to atMost currently queued messages.

func (Channel[T]) AtMost

func (c Channel[T]) AtMost(count int, options ...ConditionOption) Condition

AtMost returns a non-blocking Condition consuming up to count messages when its Wait completes.

func (Channel[T]) ChannelName

func (c Channel[T]) ChannelName() string

ChannelName returns the stable name sent to Dex and used in PersistenceSchema.

func (Channel[T]) Delete

func (c Channel[T]) Delete(ctx Context, messageID string) error

Delete stages one pending-message deletion from a Step, timeout handler, or RPC. Step deletions are best-effort. Use transactional RPCs when a missing ID must abort all writes.

func (Channel[T]) FindPendingMessage

func (c Channel[T]) FindPendingMessage(ctx Context, messageID string) (ChannelMessage[T], bool, error)

FindPendingMessage returns one loaded message by ID and reports whether it exists.

func (Channel[T]) ForN

func (c Channel[T]) ForN(count int, options ...ConditionOption) Condition

ForN returns a Condition that consumes exactly count queued messages.

func (Channel[T]) ForOne

func (c Channel[T]) ForOne(options ...ConditionOption) Condition

ForOne returns a Condition that consumes exactly one queued message.

func (Channel[T]) GetConditionResults

func (c Channel[T]) GetConditionResults(ctx Context) ([]T, error)

GetConditionResults decodes messages consumed by this Channel's satisfied Condition. It returns an error for an invalid context or an undecodable message.

func (Channel[T]) PendingMessages

func (c Channel[T]) PendingMessages(ctx Context) ([]ChannelMessage[T], error)

PendingMessages returns the loaded handler snapshot in FIFO order. It returns ChannelMessagesNotLoadedError when invocation options omitted this Channel.

func (Channel[T]) Publish

func (c Channel[T]) Publish(ctx Context, value T) error

Publish stages one message from the current Step or RPC invocation. It returns an error when ctx is invalid or value cannot be encoded.

func (Channel[T]) Size

func (c Channel[T]) Size(ctx Context) int

Size returns the invocation snapshot's queued-message count. It panics when ctx did not originate from Dex.

type ChannelDef

ChannelDef is the interface of Channel, without Go's generic

ChannelDef is internal to the SDK. Applications create values with DefineChannel or DefineChannelMap, then pass them to PersistenceSchema and Client methods.

type ChannelDef interface {
    // contains filtered or unexported methods
}

type ChannelMap

ChannelMap defines independently queued Channel instances under one shared name. Register the map once in PersistenceSchema and supply an instance string to each operation. Slash is prohibited in instance keys because it is a reserved character.

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

func DefineChannelMap

func DefineChannelMap[T any](name string) ChannelMap[T]

DefineChannelMap creates a typed Channel map with a stable name. "/" is reserved as the ChannelMap separator and is prohibited in ChannelMap names.

func (ChannelMap[T]) AllInstanceKeys

func (c ChannelMap[T]) AllInstanceKeys(ctx Context) []string

AllInstanceKeys returns non-empty instance keys in ascending order for the current handler.

func (ChannelMap[T]) AtLeast

func (c ChannelMap[T]) AtLeast(instance string, count int, options ...ConditionOption) Condition

AtLeast returns an instance Condition with an inclusive lower queue-size bound.

func (ChannelMap[T]) AtLeastAtMost

func (c ChannelMap[T]) AtLeastAtMost(instance string, atLeast int, atMost int, options ...ConditionOption) Condition

AtLeastAtMost waits for atLeast instance messages, then consumes up to atMost currently queued messages.

func (ChannelMap[T]) AtMost

func (c ChannelMap[T]) AtMost(instance string, count int, options ...ConditionOption) Condition

AtMost returns a non-blocking instance Condition consuming up to count messages when its Wait completes.

func (ChannelMap[T]) ChannelName

func (c ChannelMap[T]) ChannelName() string

ChannelName returns the stable shared name of this Channel map.

func (ChannelMap[T]) Delete

func (c ChannelMap[T]) Delete(ctx Context, instance string, messageID string) error

Delete stages one pending-message deletion from a ChannelMap instance. Step deletions are best-effort. Use transactional RPCs when a missing ID must abort all writes.

func (ChannelMap[T]) FindPendingMessage

func (c ChannelMap[T]) FindPendingMessage(ctx Context, instance string, messageID string) (ChannelMessage[T], bool, error)

FindPendingMessage returns one loaded instance message by ID and reports whether it exists.

func (ChannelMap[T]) ForN

func (c ChannelMap[T]) ForN(instance string, count int, options ...ConditionOption) Condition

ForN returns an instance Condition that consumes exactly count messages.

func (ChannelMap[T]) ForOne

func (c ChannelMap[T]) ForOne(instance string, options ...ConditionOption) Condition

ForOne returns an instance Condition that consumes exactly one message.

func (ChannelMap[T]) GetConditionResults

func (c ChannelMap[T]) GetConditionResults(ctx Context, instance string) ([]T, error)

GetConditionResults decodes messages consumed by instance's satisfied Condition. It returns an error for an invalid context or an undecodable message.

func (ChannelMap[T]) LoadMessages

func (c ChannelMap[T]) LoadMessages(instance string) ChannelMapLoad

LoadMessages selects pending messages from one slash-free logical instance for a handler snapshot.

func (ChannelMap[T]) MapSize

func (c ChannelMap[T]) MapSize(ctx Context) int

MapSize returns the number of non-empty Channel instances visible to the current handler.

func (ChannelMap[T]) PendingMessages

func (c ChannelMap[T]) PendingMessages(ctx Context, instance string) ([]ChannelMessage[T], error)

PendingMessages returns one loaded instance snapshot in FIFO order. It returns ChannelMessagesNotLoadedError when invocation options omitted this ChannelMap.

func (ChannelMap[T]) Publish

func (c ChannelMap[T]) Publish(ctx Context, instance string, value T) error

Publish stages one typed message for instance from the current invocation. It returns an error when ctx is invalid or value cannot be encoded.

func (ChannelMap[T]) Size

func (c ChannelMap[T]) Size(ctx Context, instance string) int

Size returns the invocation snapshot's queue size for instance. It panics when ctx did not originate from Dex.

type ChannelMapLoad

ChannelMapLoad selects one ChannelMap instance for a handler invocation. Create values with ChannelMap.LoadMessages and place them in matching invocation options.

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

type ChannelMessage

ChannelMessage identifies one pending Channel value loaded into a handler.

MessageID is assigned by Dex when the value is published. It can be passed to Channel.Delete.

type ChannelMessage[T any] struct {
    // MessageID is the server-assigned UUIDv7 for this pending message.
    MessageID string
    // Value is the decoded Channel value.
    Value T
}

type ChannelMessageNotFoundError

ChannelMessageNotFoundError reports a pending message ID that no longer exists.

type ChannelMessageNotFoundError struct {
    // ServiceError contains the failed deletion metadata.
    *ServiceError
}

func (*ChannelMessageNotFoundError) Unwrap

func (e *ChannelMessageNotFoundError) Unwrap() error

Unwrap returns the shared service failure.

type ChannelMessagesNotLoadedError

ChannelMessagesNotLoadedError reports pending messages omitted by invocation options.

type ChannelMessagesNotLoadedError struct {
    // Name is the Channel or ChannelMap physical name.
    Name string
}

func (*ChannelMessagesNotLoadedError) Error

func (e *ChannelMessagesNotLoadedError) Error() string

Error describes the missing pending-message load.

type Client

Client calls FlowService with registered typed definitions.

A Client is safe for concurrent use. Calls use their context for cancellation and return an error after Close. The Client owns its gRPC connection; callers should close it when the application shuts down.

client, err := dex.NewClient(registry, cache, dex.ClientOptions{})
if err != nil {
	return err
}
defer client.Close()
runID, err := client.StartFlow(ctx, orderFlow, "order-42", orderInput,
	dex.StartFlowOptions{})

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

func NewClient

func NewClient(registry *Registry, cache *blobcache.Cache, options ClientOptions) (*Client, error)

NewClient constructs a Client from shared dependencies.

registry supplies the validated definitions and cache hydrates large values. The Client connects lazily to ClientOptions.FlowServiceAddress, which defaults to localhost:8801. It panics when registry or cache is nil. It returns an error when an address, FlowServiceTLS, or GrpcMaxMessageBytes is invalid, or the gRPC client cannot be created. The caller owns the returned Client and must call Close.

func (*Client) Close

func (client *Client) Close() error

Close closes the owned FlowService connection.

Close is idempotent. Calls begun after Close return an error; calls already in progress are controlled by their contexts and the gRPC connection shutdown.

func (*Client) DescribeFlow

func (client *Client) DescribeFlow(ctx context.Context, flowID string) (FlowInfo, error)

DescribeFlow reads the current or latest run of an existing Flow without waiting.

Use it for a one-off status read, such as after a read-only RPC returns a non-terminal business state. ctx controls cancellation and flowID must be non-empty. The result includes the run identity, Flow type, status, start time, and close time when available. A missing Flow returns FlowNotFoundError; validation, context, transport, and server failures are also returned.

info, err := client.DescribeFlow(ctx, "order-123")

func (*Client) GetAttributes

func (client *Client) GetAttributes(ctx context.Context, flowID string, reads ...AttributeRead) (AttributeReadResult, error)

GetAttributes reads committed Attribute values from the current or latest Flow run.

Use GetAttributes to read Flow state from outside the Flow without defining an RPC. Pass one AttributeRead per value, created with ReadAttribute or ReadAttributeMapInstance. The Client sends all keys in one server request and does not dispatch a Worker. The read targets the current run, or the latest run when no run is open. A closed run stays readable until it is purged. The call sees only committed state. It does not see writes staged by an RPC that is still running.

The servers resolve HardDeleteOnNull values before they respond. The Client then resolves large-value references through LoadBlobs and decodes every value. GetAttributes writes destinations only after every value decodes. On error, no destination has been written. An unset Attribute or AttributeMap instance leaves its destination unchanged. The returned AttributeReadResult reports which values were set. Most callers ignore it.

A definition from any Flow in the Client's Registry is accepted. The server does not check that the target Flow's type declares it. A same-named Attribute of another type in the target Flow fails to decode.

type Note struct {
	Text string `json:"text"`
}
orderStatus := dex.DefineAttribute[string]("order-status")
notes := dex.DefineAttributeMap[Note]("notes")

var status string
var note Note
result, err := client.GetAttributes(ctx, "order-123",
	dex.ReadAttribute(orderStatus, &status),
	dex.ReadAttributeMapInstance(notes, "n1", &note),
)
var notFound *dex.FlowNotFoundError
if errors.As(err, &notFound) {
	return fmt.Errorf("order %q does not exist", notFound.FlowID)
}
if err != nil {
	return err
}
if !result.IsSet(orderStatus) {
	status = "new"
}
if !result.IsMapInstanceSet(notes, "n1") {
	note = Note{Text: "no note yet"}
}

Parameters:

Returns an AttributeReadResult and nil after every destination of a set value has been written. Returns the zero AttributeReadResult and an error otherwise:

Every error names the Flow ID. An error about one read or returned value also names its Attribute or physical key. Match error types with errors.As.

func (*Client) HealthCheck

func (client *Client) HealthCheck(ctx context.Context) (HealthInfo, error)

HealthCheck returns the responding Dex server's current health condition.

The result includes the server hostname and reported duration in milliseconds. HealthCheck performs network I/O and honors ctx. It returns context, transport, or server errors when no valid response is available.

func (*Client) InvokeRPC

func (client *Client) InvokeRPC(ctx context.Context, flowID string, rpc any, input any, outputPtr any) error

InvokeRPC synchronously invokes a registered RPC on a Flow execution.

rpc identifies an RPC definition belonging to the Flow type, input must match its input type, and outputPtr may be nil to discard the RPC output. Otherwise outputPtr must be a non-nil pointer to the RPC output type. The RPC's registered RPCOptions control timeout, locks, transactional execution, and selective state loading. A non-transactional RPC without Attribute locks starts from a backend query. If its handler returns no durable effects, a retained terminal execution can serve the query. Locks, transactional execution, returned effects, or server policy can require an active execution and cause FlowNotActiveOrNotFoundError for a terminal Flow. A missing execution also returns FlowNotActiveOrNotFoundError, including on the query-only path. InvokeRPC blocks until the handler returns, the timeout expires, or ctx is canceled, then decodes the result into outputPtr when one is provided. It may return validation, serialization, lock-conflict, worker, inactive-Flow, context, hydration, transport, or server errors. outputPtr is not owned by Dex.

var result Quote
err := client.InvokeRPC(ctx, "order-42", quoteRPC, request, &result)

func (*Client) InvokeRPCWithOptions

func (client *Client) InvokeRPCWithOptions(ctx context.Context, flowID string, rpc any, input any, outputPtr any, options RPCInvokeOptions) error

InvokeRPCWithOptions synchronously invokes a registered RPC with runtime-selected map instances.

options adds exact AttributeMap locks, AttributeMap loads, and ChannelMap loads to the immutable RPCOptions registered with the handler. Duplicate selections are sent once. Invocation options cannot remove registered state requirements. A dynamic AttributeMap read-modify-write must lock and load the same instance.

The input, output, blocking, and error behavior otherwise matches InvokeRPC.

options := dex.RPCInvokeOptions{
	LockAttributeMapInstances: []dex.AttributeLock{
		dex.LockAttributeMap(OrdersByTenant, tenantID),
	},
	LoadAttributeMapInstances: []dex.AttributeMapLoad{
		OrdersByTenant.Load(tenantID),
	},
}
err := client.InvokeRPCWithOptions(ctx, flowID, updateRPC, input, &output, options)

func (*Client) ListStreamMessages

func (client *Client) ListStreamMessages(ctx context.Context, flowID string, stream StreamDef, pageSize int32, beforePageToken string, pagePtr any) error

ListStreamMessages decodes one newest-first page of retained messages into pagePtr.

pagePtr must point to StreamMessagesPage[T] for the Stream's value type. pageSize must be positive and no greater than the server's configured limit. An empty beforePageToken starts at the retained tail. Otherwise pass StreamMessagesPage.NextPageToken unchanged to read older messages. The call does not wait for new messages. Stream trimming may remove messages between pages.

var page dex.StreamMessagesPage[string]
err := client.ListStreamMessages(ctx, "flow-1", Thinking, 100, "", &page)

func (*Client) ReadStream

func (client *Client) ReadStream(ctx context.Context, flowID string, stream StreamDef, resumeToken string, valuePtr any) (StreamMessage, error)

ReadStream returns the next retained message after resumeToken and decodes it into valuePtr.

An empty token starts at the current retained head. A token older than that head also returns the current head. The call blocks until a message arrives, the server's long-poll cap expires, or ctx is canceled. Use context.WithTimeout to impose a shorter Go-side deadline. Pass the returned StreamMessage.ResumeToken unchanged to resume after this message.

func (*Client) SearchFlows

func (client *Client) SearchFlows(ctx context.Context, query string, pageSize int32, nextPageToken string) (SearchFlowsPage, error)

SearchFlows returns one server-ordered page of Flow executions matching query.

query uses the Dex visibility query language. pageSize must satisfy the server's accepted range, and nextPageToken must be empty for the first page or copied unchanged from a previous SearchFlowsPage. Indexed attributes are hydrated before return. An empty NextPageToken marks the final page. Validation, context, hydration, transport, and server failures are returned as errors.

func (*Client) SearchFlowsWithOptions

func (client *Client) SearchFlowsWithOptions(ctx context.Context, query string, pageSize int32, nextPageToken string, options SearchFlowsOptions) (SearchFlowsPage, error)

SearchFlowsWithOptions searches Flow runs, optionally including earlier Continue-as-New runs.

func (*Client) SkipTimer

func (client *Client) SkipTimer(ctx context.Context, flowID string, stepExecution StepExecutionID, timer TimerID) error

SkipTimer makes one waiting Timer condition immediately ready.

stepExecution identifies the Step type and execution number; a nil execution number means the first execution. timer must specify exactly one condition ID or zero-based condition index. The call affects only an active Flow and returns after the server accepts the skip. Invalid identifiers, inactive Flows, context, transport, and server failures are returned as errors.

func (*Client) StartFlow

func (client *Client) StartFlow(ctx context.Context, flow Flow, flowID string, input any, options StartFlowOptions) (runID string, err error)

StartFlow starts a new Flow execution and returns its server-assigned run ID.

flow must be registered with the Client's Registry, flowID must be non-empty, and input must match the Flow's starting Step input type. options controls the request ID, timeout, initial attributes, and Flow configuration; omitted values use the registered Flow defaults. The method returns after the server accepts the Flow, not after the Flow completes.

StartFlow returns FlowAlreadyStartedError when flowID identifies an existing Flow. It also returns validation, serialization, context, transport, or server errors without transferring ownership of input. When options set AlreadyStarted.IgnoreError and a RequestID matching the existing run's start request, StartFlow returns that run ID instead of FlowAlreadyStartedError.

func (*Client) StopFlow

func (client *Client) StopFlow(ctx context.Context, flowID string, options StopOptions) error

StopFlow requests cancellation or termination of the active Flow identified by flowID.

options selects the stop mode and optional reason. The method returns after the server accepts the request; it does not wait for the Flow to close. It returns FlowNotActiveOrNotFoundError when no active execution exists, plus validation, context, transport, or server errors.

func (*Client) TimeTravel

func (client *Client) TimeTravel(ctx context.Context, flowID string, options TimeTravelOptions) (newRunID string, err error)

TimeTravel creates a new run from a selected point in an existing Flow's history.

The new run keeps the Flow ID. Writes after the selected point are reapplied unless SkipWritesReapply is true. The returned run ID identifies the new execution.

Example:

newRunID, err := client.TimeTravel(ctx, "order-123", dex.TimeTravelOptions{
	Type:            dex.TimeTravelByStepExecutionID,
	StepExecutionID: "ChargeOrder-2",
	StepMethod:      dex.TimeTravelStepExecute,
	Reason:          "retry after operator review",
})

TimeTravel returns validation errors before sending the request. It returns context, transport, Flow-not-found, or server errors after the request is sent.

func (*Client) TriggerContinueAsNew

func (client *Client) TriggerContinueAsNew(ctx context.Context, flowID string) error

TriggerContinueAsNew asks an active Flow to roll its history into a new run.

The request returns after the server accepts it and does not wait for the rollover. The Flow ID is preserved while the run ID changes. Empty identifiers, inactive Flows, context cancellation, transport failures, and server failures return errors.

func (*Client) UpdateFlowConfig

func (client *Client) UpdateFlowConfig(ctx context.Context, flowID string, config FlowConfig) error

UpdateFlowConfig updates the supplied mutable configuration fields of an active Flow.

config is validated and serialized before the request. Omitted fields keep their existing values, including the Worker target. The update applies to later Flow decisions; work already dispatched is not recalled. The method returns validation, inactive-Flow, context, transport, or server errors.

func (*Client) WaitForAttributeMapInstanceMatch

func (client *Client) WaitForAttributeMapInstanceMatch(ctx context.Context, flowID string, attribute AttributeDef, instance string, match AttributeMatchDef, valuePtr any, options WaitForAttributeOptions) error

WaitForAttributeMapInstanceMatch blocks until one AttributeMap instance satisfies match.

instance identifies the map entry. Slash is prohibited because it is reserved. The matched current value is decoded into valuePtr. The match operand and valuePtr must use the registered AttributeMap value type. options has the same required RequestID, wait-budget, and error behavior as WaitForAttributeMatch.

func (*Client) WaitForAttributeMatch

func (client *Client) WaitForAttributeMatch(ctx context.Context, flowID string, attribute AttributeDef, match AttributeMatchDef, valuePtr any, options WaitForAttributeOptions) error

WaitForAttributeMatch blocks until a singleton Attribute satisfies match.

match must contain an operand matching the registered Attribute type. The matched current value is decoded into valuePtr before this method returns. valuePtr must be a non-nil pointer of the registered type. When options.RequestID is empty, the server derives one from the Attribute condition. RequestTimeout bounds the complete call across transparent transport reattachments. A positive value returns RequestTimeoutError. InternalHandlerTimeout reclaims accepted waits that outlive callers and could consume Temporal's in-flight Update limit. Active callers transparently roll to a new generation, which adds another Update to history. Use context.WithTimeout or context.WithDeadline to bound the caller-visible response.

func (*Client) WaitForFlow

func (client *Client) WaitForFlow(ctx context.Context, flowID string, options WaitForFlowOptions) (FlowResult, error)

WaitForFlow blocks until a Flow closes or ctx ends.

options controls whether Step completion results are included. The server may end one long poll at its configured cap and return LongPollTimeoutError; callers may repeat the call. Use context.WithTimeout or context.WithDeadline for a shorter Go-side wait.

func (*Client) WaitForStepCompletion

func (client *Client) WaitForStepCompletion(ctx context.Context, flowID string, stepExecution StepExecutionID, options WaitForStepCompletionOptions) error

WaitForStepCompletion blocks until a Step execution completes, the caller-visible request expires, or ctx ends.

stepExecution identifies the Step type and execution number; nil means execution one. A nil error means the requested execution completed, but this method does not return its output. When options.RequestID is empty, the server derives a stable RequestID from the Step execution. RequestTimeout bounds the complete call across transparent transport reattachments. A positive value returns RequestTimeoutError. InternalHandlerTimeout reclaims accepted waits that outlive callers and could consume Temporal's in-flight Update limit. Active callers transparently roll to a new generation, which adds another Update to history. Invalid identifiers, inactive Flows, context, transport, and server errors are also returned.

func (*Client) WriteStream

func (client *Client) WriteStream(ctx context.Context, flowID string, stream StreamDef, source string, value any) error

WriteStream appends one best-effort message with client-supplied source metadata.

stream must be registered in exactly one Flow schema in this Client's Registry. flowID need not identify an existing or active Flow. source must be non-empty, may contain "#", and may be reused; every successful call appends a distinct message.

type ClientOptions

ClientOptions configures the FlowService client.

type ClientOptions struct {
    // FlowServiceAddress is Dex's gRPC target. Default: "localhost:8801".
    // With FlowServiceTLS, use host:port, dns:///host:port, or dns://authority/host:port, such as "dex.example.com:443".
    FlowServiceAddress string
    // FlowServiceTLS enables TLS for this Client's FlowService connection. Nil uses plaintext.
    // See FlowServiceTLSOptions for trust roots, server name, and a complete example.
    FlowServiceTLS *FlowServiceTLSOptions
    // Logger defaults to the shared BlobCache logger.
    Logger Logger
    // Credential is sent as a bearer credential with every FlowService call. Default: nil sends none.
    // A Dex server with authentication enabled requires it; see FileCredential.
    Credential Credential
    // GrpcMaxMessageBytes limits the gRPC messages this Client receives, in bytes, such as
    // InvokeRPC results, state reads, and each LoadBlobs response used to hydrate values.
    // Zero uses DefaultGrpcMaxMessageBytes (32 MiB) and keeps gRPC's unlimited send limit.
    // A positive value also limits the messages the Client sends. Negative values are invalid.
    // The Dex server applies its own api.grpcMaxMessageBytes, and the smaller limit wins.
    // Keep it above the largest application value plus headroom for the message envelope.
    GrpcMaxMessageBytes int
}

type CloseDecision

CloseDecision stores the terminal output, failure reason, and guarded Channels. Applications create it through decision factories instead of constructing it directly.

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

type Condition

Condition represents one durable Timer, Channel, or SubFlow predicate. Create values through Timer, SubFlow, Channel, or ChannelMap; custom implementations are not supported.

type Condition interface {
    // contains filtered or unexported methods
}

func SubFlow

func SubFlow(target Flow, input any, options ...SubFlowOptions) Condition

SubFlow returns a Condition satisfied when target reaches a terminal state.

target must be registered by the Worker and define a starting Step whose input accepts input. Dex generates the SubFlow Flow ID. Omit options to use the default abnormal-restart policy, or pass exactly one value to configure reuse, timing, Attributes, Flow configuration, and ID.

return dex.Until(dex.SubFlow(ChargeFlow{}, chargeInput)), nil

func Timer

func Timer(duration time.Duration, options ...ConditionOption) Condition

Timer returns a Condition satisfied after duration of durable workflow time. The timer survives Worker restarts.

type ConditionCombination

ConditionCombination groups Conditions that must all succeed as one AnyComboOf branch.

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

func Combo

func Combo(conditions ...Condition) ConditionCombination

Combo creates one branch satisfied only when all conditions are satisfied.

type ConditionOption

ConditionOption configures a Condition. Use WithConditionID to create an option.

type ConditionOption interface {
    // contains filtered or unexported methods
}

func WithConditionID

func WithConditionID(conditionID string) ConditionOption

WithConditionID assigns a stable timer or Channel Condition ID for later targeting.

type Context

Context provides invocation metadata and stages durable mutations for Step and RPC handlers.

A Context belongs to one handler attempt and must not outlive it. Attribute writes, Channel publications, locals, and events commit atomically only when the handler returns successfully.

type Context interface {
    // Context supports cancellation and deadlines for the active handler attempt.
    context.Context

    // FlowID returns the application-assigned Flow ID.
    FlowID() string
    // RunID returns the server-assigned run ID.
    RunID() string
    // FlowStartedAt returns when the current Flow run started.
    FlowStartedAt() time.Time
    // StepExecutionID returns the current Step execution ID, or empty for an RPC.
    StepExecutionID() string
    // FromStepExecutionID returns the predecessor Step execution ID, or empty when absent.
    FromStepExecutionID() string
    // RecoveryError returns the final failure that selected this recovery path, or nil when absent.
    RecoveryError() *RecoveryErrorInfo
    // FirstAttemptAt returns when the first attempt of this handler invocation started.
    FirstAttemptAt() time.Time
    // Attempt returns the one-based handler attempt number.
    Attempt() int32

    // HasTimerFired reports whether any Timer Condition completed for this Execute invocation.
    HasTimerFired() bool
    // HasTimerFiredByIndex reports whether the zero-based Timer Condition completed.
    HasTimerFiredByIndex(index int) bool
    // WaitForMethodFailed reports whether Execute followed an exhausted WaitFor with Proceed policy.
    WaitForMethodFailed() bool

    // RecordHeartbeat reports progress for the active WaitFor, Execute, or Flow timeout invocation.
    //
    // A nil value, including a typed nil, clears persisted heartbeat details. Any
    // other value is encoded and supplied to the next regular activity attempt.
    // The call waits only for the Worker response frame to enter the gRPC stream;
    // it does not wait for backend heartbeat persistence. RecordHeartbeat and
    // Stream.Write may be called concurrently and serialize their output.
    // Encoding failures return ValueMappingError; transport failures return the
    // underlying gRPC error.
    //
    // RPC invocations and calls after the handler returns produce an error.
    //
    //	var checkpoint ImportCheckpoint
    //	found, err := ctx.GetLastHeartbeatValue(&checkpoint)
    //	if err != nil {
    //		return nil, err
    //	}
    //	if !found {
    //		checkpoint = ImportCheckpoint{}
    //	}
    //	if err := ctx.RecordHeartbeat(checkpoint); err != nil {
    //		return nil, err
    //	}
    RecordHeartbeat(value any) error
    // GetLastHeartbeatValue decodes the latest retryable handler heartbeat details.
    //
    // valuePtr must be a non-nil pointer. The method returns false without changing
    // valuePtr when no details exist. Local-activity heartbeats are not available to
    // fallback regular attempts. Decode failures return ValueMappingError. See
    // RecordHeartbeat for an example.
    GetLastHeartbeatValue(valuePtr any) (found bool, err error)

    // SetStepExecutionLocal stages value under key for later attempts of this Step execution.
    SetStepExecutionLocal(key string, value any) error
    // GetStepExecutionLocal decodes a local value into non-nil valuePtr and reports whether it exists.
    GetStepExecutionLocal(key string, valuePtr any) (found bool, err error)
    // RecordEvent records one uniquely named diagnostic event with a serializable payload.
    RecordEvent(name string, value any) error
}

type Credential

Credential supplies the bearer credential that a Client or Worker sends with every FlowService call.

A Dex server with authentication enabled requires a credential on every public API call. Use FileCredential for a token that another process, such as a Kubernetes projected service account token, writes and rotates. Use APIKeyCredential when the application holds a long-lived API key that a token endpoint exchanges for short-lived access tokens.

credential, err := dex.FileCredential("/var/run/secrets/dex/token")
if err != nil {
	return err
}
client, err := dex.NewClient(registry, cache, dex.ClientOptions{Credential: credential})

The credential is sent over the same connection as every other call. Dex may be served over plaintext gRPC. Set ClientOptions.FlowServiceTLS or WorkerOptions.FlowServiceTLS to encrypt the FlowService connection that carries the credential.

type Credential interface {
    // Token returns the current bearer credential. It is called before every FlowService call,
    // may be called concurrently, and must return quickly.
    //
    // ctx is the context of the call. An error fails the call without contacting Dex.
    Token(ctx context.Context) (string, error)
}

func APIKeyCredential

func APIKeyCredential(tokenURL string, serverID string, apiKey string, options ...APIKeyCredentialOption) (Credential, error)

APIKeyCredential returns a Credential that exchanges a long-lived API key for short-lived access tokens.

Use it when a Dex server with JWT authentication issues access tokens through a token endpoint and the application holds an API key. Use FileCredential instead when another process writes and rotates the access token in a file. Pass the returned Credential in ClientOptions.Credential or WorkerOptions.Credential.

apiKey := os.Getenv("DEX_API_KEY")
credential, err := dex.APIKeyCredential(
	"https://auth.example.com/oauth/token",
	"orders-production",
	apiKey,
)
if err != nil {
	return err
}
client, err := dex.NewClient(registry, cache, dex.ClientOptions{Credential: credential})
if err != nil {
	return err
}
defer client.Close()

Exchange

An exchange sends POST to the token URL with Content-Type and Accept set to application/json. The body is {"server_id":"<server ID>","api_key":"<API key>"}. A successful response is HTTP 200 with a JSON object that holds a non-empty access_token string, a token_type equal to Bearer in any letter case, and a positive integer expires_in in seconds. Unknown response fields are ignored. Construction validates its arguments without contacting the token endpoint. The first Token call performs the first exchange.

Caching and refresh

The credential caches the access token. Its expiry is measured with the monotonic clock from the moment the exchange request started, plus expires_in. After 80% of that lifetime, the next Token call starts a refresh on a separate goroutine and still returns the cached token immediately. At most one exchange is in flight per credential, and concurrent callers that need a token share its result. When a refresh fails while the cached token is still valid, the credential keeps serving the cached token and retries the refresh at most once every 5 seconds. After the token expires, a Dex call waits for the in-flight exchange, bounded by the exchange timeout and by the call's context. The call fails with the exchange error when no token arrives.

Rejected tokens

When Dex rejects a call with gRPC UNAUTHENTICATED, the SDK discards the token that the call used, but only while it is still the cached token. A rejection of an older token therefore never discards a newer one. A Client or Worker retries a rejected unary call exactly once with a fresh token. The retry is safe because Dex rejects an unauthenticated call before executing it. A second rejection fails the call with the UNAUTHENTICATED error. The SDK observes a rejection of a pull Worker poll stream when the stream opens, when a frame is sent or received, and when sending is closed. A rejected poll stream discards its token, and the Worker reconnects once immediately with a fresh token. When that reconnection is rejected too, the Worker falls back to its usual reconnection, which waits about a minute after an UNAUTHENTICATED rejection. FileCredential and application Credential implementations get neither the unary retry nor the immediate reconnection.

Security

The token URL must use https. Plain http is allowed only for loopback hosts: localhost, 127.0.0.0/8, and ::1. The credential never follows redirects. It never includes the API key or an access token in an error, in log output, or in its formatted value. Format verbs such as %v, %+v, and %#v print only the token URL and the server ID. Errors include the token-endpoint error and error_description string fields, truncated to 200 characters, but never the raw response body. Before an error stores endpoint fields or the text of a transport error, the credential replaces each known secret with [REDACTED]. Known secrets are the API key, the two most recently issued access tokens even after they are discarded, and an access_token string in the failed response. The credential then replaces every JWT-shaped substring with [REDACTED] as well. A JWT-shaped substring is three base64url segments separated by dots, each at least 10 characters long. This rule redacts an older access token that a long-running call still uses but the credential no longer remembers. It can also redact other text of that shape, such as a long dotted host name. The Dex connection itself is plaintext gRPC, so protect the network path between the application and Dex.

Concurrency and lifecycle

The credential is safe for concurrent use and may be shared by several Clients and Workers. The application owns the credential. Closing a Client or Worker does not stop the credential's work, because other Clients and Workers may still use it. The credential starts no long-lived goroutine. Each exchange runs on its own goroutine, and that goroutine ends within one exchange timeout. The bound holds when the HTTP client's transport honors request context cancellation, as http.Transport and the default client do. The default HTTP client uses the shared http.DefaultTransport. There is nothing to close, and dropping the credential leaves no running work beyond an exchange already in flight.

Errors

APIKeyCredential returns an error when tokenURL, serverID, or apiKey is empty, when tokenURL does not parse as an absolute URL, when tokenURL uses a scheme other than https or uses http with a non-loopback host, when the exchange timeout is not positive, or when the HTTP client option is nil.

A failed exchange fails the Dex call that needed a token with an error that names the token URL. When the endpoint responded, the error also names the HTTP status and includes the error and error_description string fields when present. Exchange errors cover non-200 statuses including redirects, responses that are not a JSON object, missing or invalid access_token, token_type, or expires_in fields, transport failures, and timeouts. An exchange whose access token already expired when the response arrived also fails, because its expiry counts from the start of the request. The credential never caches or returns such a token. Transport failures and timeouts match the underlying error through errors.Is, so errors.Is(err, context.DeadlineExceeded) reports a timeout. The error text of the underlying error is redacted, and errors.As cannot extract the unredacted error. A failed refresh does not fail any call while the cached token is still valid.

tokenURL is the token endpoint. serverID identifies the Dex server that the endpoint scopes the access token to; the application supplies it and the SDK never derives it from a host name. apiKey is the secret API key. options holds optional settings. The returned Credential is safe for concurrent use.

func FileCredential

func FileCredential(path string) (Credential, error)

FileCredential returns a Credential that reads a bearer token from path.

The file is read at construction and reread whenever its modification time or size changes, checked at most once per second, so a rotated token takes effect without a restart. Surrounding whitespace is ignored. A file that becomes empty or unreadable fails the calls that need it until it is restored, rather than sending a stale token.

path is the token file. It returns an error when the file cannot be read or holds no token.

type DefaultStepOptions

DefaultStepOptions embeds GetStepOptions that returns nil (server defaults).

Example:

type ApproveOrderStep struct {
	dex.DefaultStepOptions
}

type DefaultStepOptions struct {
    // DefaultStepType supplies GetStepType using the Go type name without its package.
    DefaultStepType
}

func (DefaultStepOptions) GetStepOptions

func (DefaultStepOptions) GetStepOptions() *StepOptions

GetStepOptions returns nil so Dex uses server Step defaults.

type DefaultStepType

DefaultStepType uses the Go type name without its package as the durable step type.

type DefaultStepType struct{}

func (DefaultStepType) GetStepType

func (DefaultStepType) GetStepType() string

GetStepType returns empty so Registry derives the Go type name without its package.

type ErrorSubStatus

ErrorSubStatus is Dex's diagnostic service error classification.

type ErrorSubStatus uint8


const (
    // ErrorSubStatusUncategorized means Dex returned no specific status.
    ErrorSubStatusUncategorized ErrorSubStatus = iota + 1
    // ErrorSubStatusFlowAlreadyStarted identifies a Flow ID reuse conflict.
    ErrorSubStatusFlowAlreadyStarted
    // ErrorSubStatusFlowNotFound identifies an unknown or inactive Flow.
    ErrorSubStatusFlowNotFound
    // ErrorSubStatusWorkerAPI identifies a failed Step or RPC invocation.
    ErrorSubStatusWorkerAPI
    // ErrorSubStatusLongPollTimeout identifies an active wait whose deadline elapsed.
    ErrorSubStatusLongPollTimeout
    // ErrorSubStatusChannelMessageNotFound identifies a pending message that no longer exists.
    ErrorSubStatusChannelMessageNotFound
    // ErrorSubStatusRequestTimeout identifies an expired caller-visible request budget.
    ErrorSubStatusRequestTimeout
    // ErrorSubStatusFlowClosing identifies an update-class request rejected during Flow close cleanup.
    ErrorSubStatusFlowClosing
)

type ExecuteFailure

ExecuteFailure identifies a recovery movement after Execute exhausts retries. Create it with ProceedToOnExecuteFailure.

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

func ProceedToOnExecuteFailure

func ProceedToOnExecuteFailure[IN any](step Step[IN], options *StepOptions) *ExecuteFailure

ProceedToOnExecuteFailure routes an Execute failure to step with the same typed input. options override the recovery Step configuration for this movement; nil uses registered options.

type Flow

Flow is the top-level durable workflow definition. Any long-lived business object (at least a few seconds) can be modeled as a Flow: a Go type, a list of Steps, and a persistence schema of attributes and channels.

RPC methods are registered explicitly through GetRPCs under their Go method names.

Example:

type OrderFlow struct {
	dex.FlowDefaults
}

func (OrderFlow) GetSteps() []dex.StepDef {
	return []dex.StepDef{
		dex.DefineStartStep(ApproveOrder),
		dex.DefineStep(ShipOrder),
	}
}

func (OrderFlow) GetPersistenceSchema() dex.PersistenceSchema {
	return dex.PersistenceSchema{
		Attributes: []dex.AttributeDef{OrderStatus},
		Channels:   []dex.ChannelDef{ApprovalChannel},
	}
}

func (flow OrderFlow) GetRPCs() []dex.RPCDef {
	return []dex.RPCDef{
		dex.DefineRPC(flow.GetSnapshot, nil),
	}
}

func (OrderFlow) GetSnapshot(
	ctx dex.Context,
	input GetSnapshotInput,
) (*dex.RPCResult[OrderSnapshot], error) {
	return &dex.RPCResult[OrderSnapshot]{Output: OrderSnapshot{}}, nil
}

var Orders = OrderFlow{}
var _ dex.Flow = Orders

type Flow interface {
    // GetFlowType overrides the default Go type name without its package.
    // Embed FlowDefaults to use the default.
    //
    // The Flow type is the durable identity stored on every run. Dex Web, run
    // history, and FlowType search queries show it as is. The default for
    // *orders.OrderFlow is "OrderFlow". Name the Go type after its domain, such
    // as OrderFlow or ApprovalFlow, and never just Flow. Flow types must be
    // unique within one Registry, and the package does not disambiguate them.
    //
    // Prefer the default name. Return a non-empty compile-time string only to
    // keep an existing durable name after renaming the Go type, or to name a
    // generic Flow type, which has no default. Returning an empty string selects
    // the default.
    //
    //	func (OrderFlow) GetFlowType() string {
    //		return "OrderFulfillment"
    //	}
    GetFlowType() string

    // GetSteps defines the steps of the flow. A step is one node in the flow
    // state machine: it may WaitFor conditions (channel / timer) and then
    // Execute a decision. See Step for details.
    //
    // Use DefineStartStep for at most one starting step and DefineStep for
    // every other step. An empty list, or a list with no starting step, means
    // the run starts with no step execution; application code can still invoke
    // RPCs that move into steps later.
    GetSteps() []StepDef

    // GetRPCs defines the Flow's synchronous RPC methods and their immutable execution options.
    // Embed FlowDefaults to return an empty list when the Flow exposes no RPCs.
    GetRPCs() []RPCDef

    // GetPersistenceSchema declares attributes and channels for this flow.
    //
    // Attributes can be read and written in WaitFor, Execute, and RPC. External
    // clients can also read them through Client APIs. Indexed attributes can be
    // used in SearchFlows queries.
    //
    // Channels carry messages into a waiting step: external clients publish;
    // WaitFor requests consumption and Execute reads results via
    // Channel.GetConditionResults. Channel maps key messages by string.
    GetPersistenceSchema() PersistenceSchema
}

type FlowAlreadyStartedError

FlowAlreadyStartedError reports a duplicate Flow start.

type FlowAlreadyStartedError struct {
    // ServiceError contains the failed StartFlow metadata.
    *ServiceError
}

func (*FlowAlreadyStartedError) Unwrap

func (e *FlowAlreadyStartedError) Unwrap() error

Unwrap returns the shared service failure.

type FlowClosingError

FlowClosingError reports that a request arrived while Dex was closing a Flow. Re-read the retained terminal Flow state before deciding whether to retry or take another action. Match it with errors.As to avoid depending on the service error text:

import (
	"context"
	"errors"
	"github.com/superdurable/dex/sdk-go/dex"
)

func readTerminalFlowAfterClosing(ctx context.Context, client *dex.Client, flowID string, callErr error) (dex.FlowResult, error) {
	var closing *dex.FlowClosingError
	if errors.As(callErr, &closing) {
		return client.WaitForFlow(ctx, flowID, dex.WaitForFlowOptions{})
	}
	return dex.FlowResult{}, callErr
}

type FlowClosingError struct {
    // ServiceError contains the rejected operation and original transport failure.
    *ServiceError
}

func (*FlowClosingError) Unwrap

func (e *FlowClosingError) Unwrap() error

Unwrap returns the shared service failure.

type FlowConfig

FlowConfig overrides mutable runtime behavior for one Flow. Nil fields preserve server defaults.

type FlowConfig struct {
    // ActiveStepSearchMode controls active Step indexing.
    ActiveStepSearchMode *ActiveStepSearchMode
    // ContinueAsNewThreshold is the event-count threshold for Continue-As-New.
    ContinueAsNewThreshold *int32
    // ContinueAsNewPageSizeBytes caps history bytes carried into Continue-As-New.
    ContinueAsNewPageSizeBytes *int32
    // StepDurability sets the default persistence durability for Step writes.
    StepDurability *StepDurability
    // AttributeStoreNames selects Server-configured Attribute Stores for opted-in writes.
    // A nil slice preserves the current targets; an empty non-nil slice disables future projections.
    AttributeStoreNames []string
}

type FlowDefaults

FlowDefaults uses the Go type name without its package as the durable flow type.

type FlowDefaults struct{}

func (FlowDefaults) GetFlowType

func (FlowDefaults) GetFlowType() string

GetFlowType returns empty so Registry derives the Go type name without its package.

func (FlowDefaults) GetRPCs

func (FlowDefaults) GetRPCs() []RPCDef

GetRPCs returns no RPC definitions. Flows exposing RPCs override this method.

type FlowDefinitionError

FlowDefinitionError reports an invalid or unregistered Flow definition.

type FlowDefinitionError struct {
    // FlowType identifies the invalid or unregistered Flow when known.
    FlowType string
    // Definition identifies the invalid Step, RPC, Attribute, or Channel when known.
    Definition string
    // Err preserves the developer-actionable contract violation.
    Err error
}

func (*FlowDefinitionError) Error

func (e *FlowDefinitionError) Error() string

Error formats the Flow definition and its underlying contract violation.

func (*FlowDefinitionError) Unwrap

func (e *FlowDefinitionError) Unwrap() error

Unwrap returns the underlying definition violation.

type FlowErrorType

FlowErrorType categorizes the failure recorded for a terminal Flow.

type FlowErrorType uint8


const (
    // FlowErrorStepDecision identifies an invalid or failing Step decision.
    FlowErrorStepDecision FlowErrorType = iota + 1
    // FlowErrorClientAPI identifies a client API operation that failed the Flow.
    FlowErrorClientAPI
    // FlowErrorWorkerMethod identifies a failed Worker Step or RPC handler.
    FlowErrorWorkerMethod
    // FlowErrorInvalidUserCode identifies an invalid Flow or Step definition.
    FlowErrorInvalidUserCode
    // FlowErrorInternal identifies an internal Dex failure.
    FlowErrorInternal
    // FlowErrorTimeout identifies expiration of a Dex soft Flow timeout.
    FlowErrorTimeout
)

type FlowInfo

FlowInfo describes the current or latest run for one Flow ID.

Client.DescribeFlow returns this snapshot without waiting for a terminal status. A zero ClosedAt means the run is open or its close time is unavailable.

type FlowInfo struct {
    // FlowID is the application-assigned Flow ID.
    FlowID string
    // RunID is the server-assigned run ID.
    RunID string
    // FlowType is the registered Flow type.
    FlowType string
    // Status is the current or terminal Flow status.
    Status FlowStatus
    // StartedAt is the run start time.
    StartedAt time.Time
    // ClosedAt is the run close time, if available.
    ClosedAt time.Time
}

type FlowNotActiveOrNotFoundError

FlowNotActiveOrNotFoundError reports a missing Flow or an operation that cannot use a closed Flow. InvokeRPC also returns this error when a query-only RPC finds no readable execution. A query-only RPC without locks, transactionality, durable effects, or forced Update routing can read a retained terminal execution. At that confirmed read boundary, treat this error as not found. For mutations and active-only RPCs, this error does not distinguish missing from closed Flows or prove that the requested work succeeded. Other service and Worker failures remain distinct. Use errors.As to match the error and access its ServiceError metadata:

var unavailable *dex.FlowNotActiveOrNotFoundError
if errors.As(err, &unavailable) {
	return unavailable
}

type FlowNotActiveOrNotFoundError struct {
    // ServiceError contains the failed operation's metadata and original transport cause.
    *ServiceError
}

func (*FlowNotActiveOrNotFoundError) Unwrap

func (e *FlowNotActiveOrNotFoundError) Unwrap() error

Unwrap returns the shared service failure.

type FlowNotFoundError

FlowNotFoundError reports a missing Flow for an existing-Flow operation.

type FlowNotFoundError struct {
    // ServiceError contains the failed existing-Flow operation metadata.
    *ServiceError
}

func (*FlowNotFoundError) Unwrap

func (e *FlowNotFoundError) Unwrap() error

Unwrap returns the shared service failure.

type FlowResult

FlowResult describes an observed Flow status and its output-bearing Step completions.

Client.WaitForFlow returns terminal results. SubFlowResult can return a running snapshot when another branch of AnyOf wins; that snapshot is not a live backend query.

type FlowResult struct {
    // Status is the observed Flow status.
    Status FlowStatus
    // Completions preserves server collection order when NeedsResults was true.
    // Parallel completion order is not deterministic; select by StepType or StepExecutionID.
    Completions []StepCompletion
    // ErrorType is the Flow failure category when available.
    ErrorType FlowErrorType
    // ErrorMessage is the server completion detail when available.
    ErrorMessage string
}

func SubFlowResult

func SubFlowResult(ctx Context, index ...int) (FlowResult, error)

SubFlowResult returns one SubFlow result from the current Execute invocation.

Omit index to read the first SubFlow Condition. Indexes are zero-based in the stable SubFlow order within the surrounding Wait. AnyOf losers return a nonterminal running snapshot.

func (FlowResult) DecodeSingleOutput

func (result FlowResult) DecodeSingleOutput(target any) error

DecodeSingleOutput decodes the output when exactly one completion exists.

target follows Value.Decode semantics and must be a non-nil pointer. Running results and zero or multiple completions return an error without decoding a value.

func (FlowResult) IsTerminal

func (result FlowResult) IsTerminal() bool

IsTerminal reports whether the result can no longer execute in its observed run.

type FlowRetryPolicy

FlowRetryPolicy configures whole-Flow retry timing after terminal failure.

type FlowRetryPolicy struct {
    // InitialInterval is the delay before the first Flow retry.
    InitialInterval time.Duration
    // BackoffCoefficient multiplies successive Flow retry intervals.
    BackoffCoefficient float64
    // MaximumInterval caps time between Flow retries.
    MaximumInterval time.Duration
    // MaximumAttempts limits total Flow attempts, including the initial attempt.
    MaximumAttempts int32
}

type FlowServiceTLSOptions

FlowServiceTLSOptions enables TLS for a Client or Worker's FlowService connection.

Set ClientOptions.FlowServiceTLS and WorkerOptions.FlowServiceTLS independently. A nil pointer keeps plaintext gRPC. A non-nil pointer uses TLS 1.2 or newer, verifies the server certificate, and negotiates HTTP/2 with ALPN h2. The FlowServiceAddress host is the default certificate name and SNI. Without ServerName, HTTP/2 :authority is the address host and port, as with plaintext. An IPv6 address keeps its brackets in :authority. The TLS settings apply to the outbound FlowService and Worker task connections. Configure the APIKeyCredential token endpoint separately. With TLS, FlowServiceAddress accepts only host:port, dns:///host:port, or dns://authority/host:port (with an explicit DNS resolver authority). The DNS resolver authority never becomes the certificate name or :authority. Other gRPC resolver schemes are available only with plaintext connections. Connections open on demand, so certificate, hostname, and handshake failures appear as Unavailable on the first Client call or from Worker.Start. The endpoint or gateway must accept ALPN h2 and forward gRPC over HTTP/2. An IP address or IP ServerName sends no SNI and requires a matching IP SAN.

func run(ctx context.Context, registry *dex.Registry, cache *blobcache.Cache) error {
	rootPEM, err := os.ReadFile("/etc/dex/root-ca.pem")
	if err != nil {
		return err
	}
	tlsOptions := &dex.FlowServiceTLSOptions{AdditionalRootCAsPEM: rootPEM}
	worker, err := dex.NewWorker(registry, cache, dex.WorkerOptions{
		FlowServiceAddress: "dex.example.com:443", FlowServiceTLS: tlsOptions,
		Pull: &dex.PullWorkerOptions{},
	})
	if err != nil {
		return err
	}
	client, err := dex.NewClient(registry, cache, dex.ClientOptions{
		FlowServiceAddress: "dex.example.com:443", FlowServiceTLS: tlsOptions,
	})
	if err != nil {
		return errors.Join(err, worker.Stop(context.Background()))
	}
	workerResult := make(chan error, 1)
	go func() { workerResult <- worker.Start() }()
	if err := worker.WaitPullRegistered(ctx); err != nil {
		return errors.Join(err, worker.Stop(context.Background()), <-workerResult, client.Close())
	}
	<-ctx.Done()
	return errors.Join(worker.Stop(context.Background()), <-workerResult, client.Close())
}

type FlowServiceTLSOptions struct {
    // AdditionalRootCAsPEM adds PEM-encoded root CAs to the system roots.
    // Empty uses the system roots alone. The bytes are read at construction, not watched for changes.
    // At least one certificate must parse from a nonempty bundle, or NewClient or NewWorker fails.
    // Non-certificate and unparsable PEM blocks are skipped when another certificate parses.
    // Without system roots or an added trusted CA, certificate verification fails on connection.
    AdditionalRootCAsPEM []byte
    // ServerName overrides the FlowServiceAddress host for certificate verification, SNI, and HTTP/2 :authority.
    // Empty uses the FlowServiceAddress host for verification and SNI, and its host and port for :authority.
    // Set it only when the certificate name differs from the address host.
    // A nonempty value is also sent as :authority, without a port.
    // A gateway that routes by :authority must accept that name.
    // Use a DNS name or IP address without a port.
    // An IP address sends no SNI and requires a matching IP SAN.
    // An IPv6 ServerName is sent as an unbracketed :authority, which strict HTTP/2 servers reject.
    // For an IPv6 endpoint, leave ServerName empty or use a DNS name that the certificate covers.
    // An invalid value fails NewClient or NewWorker.
    ServerName string
}

type FlowStatus

FlowStatus describes the current or terminal state of a Flow run.

type FlowStatus uint8


const (
    // FlowRunning means the Flow can still execute Steps or receive messages and RPCs.
    FlowRunning FlowStatus = iota + 1
    // FlowCompleted means the Flow completed successfully.
    FlowCompleted
    // FlowFailed means an unrecovered Step, RPC, or user-code error ended the Flow.
    FlowFailed
    // FlowServerSideTimeoutInternalOnly is reserved for backend hard-timeout reporting.
    // Applications must not depend on this status.
    FlowServerSideTimeoutInternalOnly
    // FlowTerminated means an operator ended the Flow immediately.
    FlowTerminated
    // FlowCanceled means the Flow completed cooperative cancellation.
    FlowCanceled
    // FlowContinuedAsNew means history rolled into a successor run.
    FlowContinuedAsNew
)

func (FlowStatus) String

func (status FlowStatus) String() string

String returns the Flow status name.

type FlowTimeoutHandler

FlowTimeoutHandler lets a Flow execute application code when its soft timeout expires.

Implement this capability on the same Flow value registered with NewRegistry. A positive StartFlowOptions.Timeout defaults to TimeoutHandler when this capability is present. The returned StepDecision has the same validation and terminal behavior as a Step Execute result. Context exposes the timed-out Flow execution and permits normal Attribute and Channel writes.

HandleTimeout may return GoTo, DeadEnd, ForceComplete, ForceFail, or GracefulComplete. It starts after Dex's durable timeout timer and may retry according to FlowTimeoutHandlerOptions.

type FlowTimeoutHandler interface {
    // HandleTimeout executes the Flow's soft-timeout hook.
    //
    // ctx belongs to this invocation and must not be retained. A non-nil error applies normal
    // Execute retry and failure policy behavior; a nil decision is rejected as invalid.
    HandleTimeout(ctx Context) (*StepDecision, error)
}

type FlowTimeoutHandlerFailure

FlowTimeoutHandlerFailure identifies a recovery Step after a timeout handler exhausts retries. Create it with ProceedToOnFlowTimeoutHandlerFailure. The target must be a Step[None] registered by the timed-out Flow. The target reads the final handler error from Context.

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

func ProceedToOnFlowTimeoutHandlerFailure

func ProceedToOnFlowTimeoutHandlerFailure(step Step[None], options *StepOptions) *FlowTimeoutHandlerFailure

ProceedToOnFlowTimeoutHandlerFailure routes a timeout handler failure to a no-input Step. options override the recovery Step configuration for this movement; nil uses registered options. The recovery Step receives None and reads the final handler error from Context.

type FlowTimeoutHandlerOptions

FlowTimeoutHandlerOptions configures timeout-handler execution and state loading. Zero values preserve the server's Execute defaults. Ordinary Attributes and Channel size metadata are loaded automatically. AttributeMap values and pending Channel messages require an explicit load selection. One logical handler execution may contain multiple retry attempts.

type FlowTimeoutHandlerOptions struct {
    // MethodTimeout limits one timeout-handler attempt; zero uses the server default.
    MethodTimeout time.Duration
    // HeartbeatTimeout detects a stalled regular timeout-handler attempt.
    // Zero uses the server default. Positive values must be whole seconds within int32 range.
    HeartbeatTimeout time.Duration
    // Retry overrides the timeout-handler retry policy; nil uses server defaults.
    Retry *RetryPolicy
    // Failure selects a no-input recovery Step after the handler exhausts retries.
    Failure *FlowTimeoutHandlerFailure
    // Durability overrides persistence durability for timeout-handler writes.
    Durability StepDurability
    // LockAttributes are acquired together for the timeout-handler invocation.
    LockAttributes []AttributeLock
    // LoadAttributeMaps includes every current instance of each AttributeMap.
    LoadAttributeMaps []AttributeDef
    // LoadAttributeMapInstances includes exact AttributeMap instances.
    LoadAttributeMapInstances []AttributeMapLoad
    // LoadChannels includes pending messages from selected Channels.
    LoadChannels []ChannelDef
    // LoadChannelMaps includes every current ChannelMap instance.
    LoadChannelMaps []ChannelDef
    // LoadChannelMapInstances includes exact ChannelMap instance messages.
    LoadChannelMapInstances []ChannelMapLoad
}

type FlowTimeoutPolicy

FlowTimeoutPolicy controls how a positive Flow timeout ends or redirects execution.

type FlowTimeoutPolicy uint8


const (
    // TimeoutDefault selects Handler for a FlowTimeoutHandler and Fail otherwise.
    TimeoutDefault FlowTimeoutPolicy = iota
    // TimeoutFail fails the Flow with FlowErrorTimeout and permits Flow retries.
    TimeoutFail
    // TimeoutCancel cancels the Flow without retrying it.
    TimeoutCancel
    // TimeoutHandler invokes FlowTimeoutHandler.HandleTimeout as one retryable logical execution.
    TimeoutHandler
)

type HardDeleteAttributeUnsupportedError

HardDeleteAttributeUnsupportedError reports a hard-delete Attribute write to a Server that cannot keep it out of history.

Attributes defined with HardDeleteOnNull require Server protocol version 3 or later. Older Servers would ignore the hard-delete marker and store the value in Flow history, so the SDK refuses the write instead. Client.StartFlow returns this error before sending initial values. A Step or RPC that writes such an Attribute fails with this error, and Dex retries it according to its retry policy. Upgrade the Dex Server to resolve it.

Example:

var unsupported *dex.HardDeleteAttributeUnsupportedError
if errors.As(err, &unsupported) {
	log.Printf("upgrade Dex: %s needs protocol %d", unsupported.AttributeName, unsupported.RequiredServerProtocolVersion)
}

type HardDeleteAttributeUnsupportedError struct {
    // AttributeName is the physical Attribute key of the refused write, including any map instance.
    AttributeName string
    // NegotiatedServerProtocol is the protocol version agreed with the Server.
    NegotiatedServerProtocol uint32
    // RequiredServerProtocolVersion is the minimum Server protocol version for hard-delete Attributes.
    RequiredServerProtocolVersion uint32
}

func (*HardDeleteAttributeUnsupportedError) Error

func (e *HardDeleteAttributeUnsupportedError) Error() string

Error describes the refused write and the required Server protocol version.

type HealthInfo

HealthInfo describes one Dex health-check condition.

type HealthInfo struct {
    // Condition is the service-reported health state.
    Condition string
    // Hostname identifies the responding Dex server.
    Hostname string
    // Duration is the response duration in milliseconds.
    Duration int32
}

type IDReusePolicy

IDReusePolicy controls whether StartFlow may reuse a Flow ID.

type IDReusePolicy uint8


const (
    // IDReuseDefault uses the server default, which rejects every previously used Flow ID.
    IDReuseDefault IDReusePolicy = iota
    // IDReuseAllowIfPreviousFailed permits reuse only after a failed run.
    IDReuseAllowIfPreviousFailed
    // IDReuseAllowIfNotRunning permits reuse when no run is active.
    IDReuseAllowIfNotRunning
    // IDReuseDisallow rejects reuse regardless of previous status.
    IDReuseDisallow
    // IDReuseTerminateIfRunning terminates an active run before replacement.
    IDReuseTerminateIfRunning
)

type IndexType

IndexType identifies the search representation required for an indexed Attribute value.

type IndexType uint8


const (
    // IndexKeyword indexes one exact UTF-8 string for equality and aggregation queries.
    IndexKeyword IndexType = iota + 1
    // IndexFullText indexes analyzed UTF-8 text for full-text queries.
    IndexFullText
    // IndexKeywordArray indexes a slice or array of exact UTF-8 strings.
    IndexKeywordArray
    // IndexInt indexes a signed or unsigned integer that fits in int64.
    IndexInt
    // IndexDouble indexes a finite floating-point value.
    IndexDouble
    // IndexBool indexes a Boolean value.
    IndexBool
    // IndexDatetime indexes time.Time or an RFC3339Nano string.
    IndexDatetime
)

type InitialAttributeDef

InitialAttributeDef is an encoded Attribute initialization accepted by StartFlowOptions. Values are created by InitialAttribute and InitialAttributeMapValue.

type InitialAttributeDef interface {
    // contains filtered or unexported methods
}

func InitialAttribute

func InitialAttribute[T any](attribute Attribute[T], value T) (InitialAttributeDef, error)

InitialAttribute encodes a value for StartFlowOptions.Attributes. It returns ValueMappingError when the value is unsupported or incompatible with its index.

func InitialAttributeMapValue

func InitialAttributeMapValue[T any](attribute AttributeMap[T], instance string, value T) (InitialAttributeDef, error)

InitialAttributeMapValue encodes one Attribute-map instance for StartFlowOptions.Attributes. The instance identifies the map entry. Slash is prohibited in instance keys because it is a reserved character. It returns ValueMappingError when the value is unsupported or incompatible with its index.

type InvalidStepResultError

InvalidStepResultError reports an invalid Worker handler result.

type InvalidStepResultError struct {
    // FlowType identifies the containing Flow.
    FlowType string
    // StepType identifies the Step, or empty for an RPC.
    StepType string
    // Method is WaitFor, Execute, or the RPC name.
    Method string
    // Err preserves the violated result contract.
    Err error
}

func (*InvalidStepResultError) Error

func (e *InvalidStepResultError) Error() string

Error formats the handler location and invalid-result detail.

func (*InvalidStepResultError) Unwrap

func (e *InvalidStepResultError) Unwrap() error

Unwrap returns the underlying result-contract violation.

type ListedStreamMessage

ListedStreamMessage contains one decoded message returned by Client.ListStreamMessages.

Values use the Stream's registered type. ResumeToken identifies this retained message. CreatedTime is assigned by the Stream Store. Source identifies the producer.

type ListedStreamMessage[T any] struct {
    // Value is the decoded application message.
    Value T
    // ResumeToken identifies this retained message.
    ResumeToken string
    // CreatedTime is the server-assigned creation time.
    CreatedTime time.Time
    // Source is client-supplied metadata or the Step's #stepExecutionID value.
    Source string
}

type Logger

Logger receives structured SDK logs.

type Logger = logging.Logger

type LongPollTimeoutError

LongPollTimeoutError reports an expired server long poll.

type LongPollTimeoutError struct {
    // ServiceError contains the timed-out wait metadata.
    *ServiceError
}

func (*LongPollTimeoutError) Unwrap

func (e *LongPollTimeoutError) Unwrap() error

Unwrap returns the shared service failure.

type NoWaitFor

NoWaitFor marks a Step as execute-only. Embed it (or StepDefaultsNoWaitFor) so registration sets skip_wait_for and never calls WaitFor. The WaitFor method exists only to satisfy Step[IN] and panics if invoked.

Example:

type ShipOrderStep struct {
	dex.NoWaitFor[ShipOrderInput]
	dex.DefaultStepOptions
}

type NoWaitFor[IN any] struct{}

func (NoWaitFor[IN]) WaitFor

func (NoWaitFor[IN]) WaitFor(Context, IN) (*Wait, error)

WaitFor panics because registration must skip this method for execute-only Steps.

type None

None is the nil-only SDK payload, rejecting arbitrary values unlike any and clarifying struct{} intent.

type None = *none

type PersistenceSchema

PersistenceSchema registers a Flow's Attribute, Channel, and Stream definitions.

type PersistenceSchema struct {
    // Attributes contains every Attribute and AttributeMap used by the Flow.
    Attributes []AttributeDef
    // Channels contains every Channel and ChannelMap used by the Flow.
    Channels []ChannelDef
    // Streams contains every best-effort Stream owned by the Flow type.
    Streams []StreamDef
}

type PullQueueRegistrationState

PullQueueRegistrationState describes one queue in a PullRegistrationState snapshot.

Inspect these entries to diagnose a failing readiness probe. A queue may have both a registered stream and LastPermanentError when another stream was rejected. The snapshot owns its values; errors are shared immutable errors. See WaitPullRegistered for the startup and readiness probe example.

type PullQueueRegistrationState struct {
    // Queue is the Flow type name or app Worker ID identifying this queue.
    Queue string
    // HasRegisteredStream reports whether at least one acknowledged stream is currently held.
    HasRegisteredStream bool
    // LastPermanentError is the last permanent gRPC status error, or nil if none remains.
    // A successful registration clears it. Transient failures do not overwrite it.
    LastPermanentError error
}

type PullRegistrationError

PullRegistrationError identifies a permanent rejection that leaves a pull queue without a registered stream.

WaitPullRegistered returns this error for such a rejection before startup readiness. Start also returns it when StopOnPermanentStreamRejection is enabled. Use errors.As to inspect Queue and status.Code on Err to inspect the gRPC status:

var rejection *dex.PullRegistrationError
if errors.As(err, &rejection) {
	log.Printf("queue %s rejected: %s", rejection.Queue, status.Code(rejection.Err))
}

SDK-produced values always have a nonempty Queue and non-nil Err. Treat returned errors as immutable; they can be shared by concurrent readiness callers.

type PullRegistrationError struct {
    // Queue names the rejected Flow type queue or app Worker ID queue.
    Queue string
    // Err is the non-nil gRPC status error returned by the rejected stream.
    Err error
}

func (*PullRegistrationError) Error

func (rejection *PullRegistrationError) Error() string

Error returns the queue name and underlying rejection for diagnostics. It does not change the error; concurrent calls are safe on immutable values.

func (*PullRegistrationError) Unwrap

func (rejection *PullRegistrationError) Unwrap() error

Unwrap returns Err so errors.Is, errors.As, and gRPC status inspection retain the rejection. SDK-produced errors return a non-nil error. See PullRegistrationError for usage.

type PullRegistrationState

PullRegistrationState is an independent snapshot for a pull Worker's readiness probe.

Obtain it with Worker.PullRegistrationState, including concurrently with Start, Stop, or WaitPullRegistered. Its zero value has IsReady false and Queues nil. Modifying Queues does not reconfigure the Worker. Treat errors inside entries as immutable. See WaitPullRegistered for a complete startup and HTTP readiness probe example.

type PullRegistrationState struct {
    // IsReady becomes true after every queue has registered at least once.
    // Afterwards it is false only if a queue has no registered stream and a permanent error.
    // Transient failures and ownership moves preserve readiness. Successful registration clears that queue's error.
    IsReady bool
    // Queues contains the app Worker ID queue followed by Flow type queues in sorted order.
    // The Worker returns a fresh non-nil slice.
    Queues []PullQueueRegistrationState
}

type PullWorkerOptions

PullWorkerOptions configures outbound polling of Dex task queues.

Use it with WorkerOptions.Pull to tune outbound task streams. MaxConcurrentTasks defaults to runtime.NumCPU; IdleStreamsPerQueue defaults to one. StopDrainTimeout defaults to 30 seconds. Each stream runs one task at a time. Negative values and an invalid AppWorkerID cause NewWorker to return an error.

worker, err := dex.NewWorker(registry, cache, dex.WorkerOptions{
	Pull: &dex.PullWorkerOptions{MaxConcurrentTasks: 8},
})
if err != nil { return err }
defer worker.Stop(context.Background())

type PullWorkerOptions struct {
    // StopOnPermanentStreamRejection makes Start return a permanent rejection when its queue has no registered stream. Default: false retries indefinitely.
    // A rejection while another stream holds the same queue is recorded and retried without stopping.
    // This applies before and after initial registration. Start begins draining all queues and bounds
    // the drain with StopDrainTimeout, then cancels streams and waits for handlers to return.
    // Handlers must respond to cancellation. Other queues may still hold streams while draining.
    // Always call Stop before exiting; it remains safe after Start returns.
    StopOnPermanentStreamRejection bool
    // MaxConcurrentTasks bounds simultaneous handlers. Zero uses runtime.NumCPU().
    MaxConcurrentTasks int
    // ReservedRPCTasks protects synchronous RPC latency from long-running Steps.
    // Nil reserves one slot when resolved MaxConcurrentTasks is at least two, otherwise zero.
    // With the CPU-count default, Steps therefore get one fewer slot than before. Set ptr.Any(0) to opt out.
    // With one slot and nil reserve, Start warns once that RPCs may wait behind Steps.
    // Flows must stay responsive because RPCs share Worker task capacity with Steps.
    // WaitFor and Execute tasks, including local attempts, are declined when total busy handlers reach
    // MaxConcurrentTasks minus the reserve. API-direct and transactional RPC tasks may use the full limit.
    // NewWorker returns an error unless 0 <= reserve < resolved MaxConcurrentTasks.
    // Declines reassign without consuming an attempt. If all Workers stay at their Step limit until the
    // dispatch wait expires, NO_WORKER_AVAILABLE consumes an attempt and follows the Step retry policy.
    // Local attempts wait the local dispatch wait, then fall back according to their policy.
    // Declines emit a Flow-type debug log, increment PullReservedRPCDeclines, and preserve readiness.
    // A queue keeps one extra idle stream for 30 seconds after its last reserve decline.
    // The Worker does not open that stream while all task slots are busy.
    // For example: PullWorkerOptions{MaxConcurrentTasks: 8, ReservedRPCTasks: ptr.Any(2)}.
    ReservedRPCTasks *int
    // IdleStreamsPerQueue sets waiting streams for each Flow type and the sticky queue. Zero uses one.
    // A queue keeps one more after a recent reserve decline. Use at least two for heavy Step backlogs.
    IdleStreamsPerQueue int
    // AppWorkerID overrides the generated process identity. Empty generates identity plus eight base32 characters.
    // Set a new value after each process restart. It must not equal a registered Flow type.
    // Valid values contain 1-63 ASCII letters, digits, dots, underscores, or hyphens.
    AppWorkerID string
    // StopDrainTimeout bounds Stop after it begins draining. Zero uses 30 seconds; negative values are invalid.
    StopDrainTimeout time.Duration
}

type RPC

RPC is a typed Flow method that may read persistence and return output plus Step movements.

Register a direct bound Flow method with DefineRPC. Dex uses the Go method name as the durable RPC name. Return a non-nil RPCResult on success or an error to report a Worker invocation failure.

Example:

func (OrderFlow) GetStatus(
	ctx dex.Context,
	input GetStatusInput,
) (*dex.RPCResult[OrderStatus], error) {
	return &dex.RPCResult[OrderStatus]{Output: OrderStatus{}}, nil
}

type RPC[IN, OUT any] func(
    ctx Context,
    input IN,
) (*RPCResult[OUT], error)

type RPCDef

RPCDef is the registration representation of a typed RPC method.

Applications create definitions with DefineRPC and return them from Flow.GetRPCs. Its methods are sealed so only SDK-provided definitions can be registered.

type RPCDef interface {
    // contains filtered or unexported methods
}

func DefineRPC

func DefineRPC[IN, OUT any](rpc RPC[IN, OUT], options *RPCOptions) RPCDef

DefineRPC registers a direct bound Flow method with immutable execution options.

Pass nil options to use server defaults without Attribute locks, transactional execution, or selective collection loading. The Registry validates the method and options before publishing the Flow definition.

func (flow *OrderFlow) GetRPCs() []dex.RPCDef {
	return []dex.RPCDef{
		dex.DefineRPC(flow.GetOrder, nil),
		dex.DefineRPC(flow.UpdateOrder, &dex.RPCOptions{
			LockAttributes: []dex.AttributeLock{dex.LockAttribute(OrderStatus)},
		}),
	}
}

type RPCInvokeOptions

RPCInvokeOptions adds runtime-selected map instances to one RPC invocation.

The Client merges these selections with the RPC's registered RPCOptions. Invocation options cannot remove registered locks or loads. AttributeMap locks and loads remain independent; a read-modify-write RPC must include the same instance in both fields.

type RPCInvokeOptions struct {
    // LockAttributeMapInstances are acquired with the RPC's registered Attribute locks.
    // Every lock must be created with LockAttributeMap.
    LockAttributeMapInstances []AttributeLock
    // LoadAttributeMapInstances adds exact AttributeMap instances to the RPC snapshot.
    LoadAttributeMapInstances []AttributeMapLoad
    // LoadChannelMapInstances adds exact ChannelMap instance messages to the RPC snapshot.
    LoadChannelMapInstances []ChannelMapLoad
}

type RPCLockConflictError

RPCLockConflictError reports concurrent RPC lock contention.

type RPCLockConflictError struct {
    // ServiceError contains the aborted RPC metadata.
    *ServiceError
}

func (*RPCLockConflictError) Unwrap

func (e *RPCLockConflictError) Unwrap() error

Unwrap returns the shared service failure.

type RPCOptions

RPCOptions configures every invocation of one registered RPC.

Return these options with DefineRPC from Flow.GetRPCs. The Client applies the same immutable settings to every invocation. Its context may still impose a shorter caller-side deadline.

type RPCOptions struct {
    // Action declares this RPC as a state-dependent Action with one required permission.
    // Nil leaves the RPC outside Work Queue permission projection. The projection identifies
    // candidate Runs only; the RPC gateway must independently authorize every invocation.
    Action ActionDef
    // Timeout limits the RPC handler; zero uses the server default.
    Timeout time.Duration
    // LockAttributes are acquired atomically for the RPC invocation.
    LockAttributes []AttributeLock
    // IsTransactional requests transactional reads and writes without requiring Attribute locks.
    // Channel deletions require this option to make a missing message abort all RPC writes.
    IsTransactional bool
    // LoadAttributeMaps includes every current instance of each AttributeMap in the RPC snapshot.
    LoadAttributeMaps []AttributeDef
    // LoadAttributeMapInstances includes exact AttributeMap instances in the RPC snapshot.
    LoadAttributeMapInstances []AttributeMapLoad
    // LoadChannels includes pending messages from the selected Channels in the RPC snapshot.
    LoadChannels []ChannelDef
    // LoadChannelMaps includes every current instance of each ChannelMap in the RPC snapshot.
    LoadChannelMaps []ChannelDef
    // LoadChannelMapInstances includes exact ChannelMap instance messages in the RPC snapshot.
    LoadChannelMapInstances []ChannelMapLoad
}

type RPCResult

RPCResult carries typed output, Step movements, and Flow-wide Step cancellation.

type RPCResult[OUT any] struct {
    // Output is encoded as the RPC response value.
    Output OUT
    // NextSteps are scheduled in order after the RPC persistence changes commit.
    NextSteps []StepMovement
    // CancelingSteps selects registered Step types canceled before NextSteps are scheduled.
    CancelingSteps []StepSelector
}

func (*RPCResult[OUT]) CancelSteps

func (result *RPCResult[OUT]) CancelSteps(steps ...StepSelector) *RPCResult[OUT]

CancelSteps selects queued or active executions of registered Step types.

Dex resolves the selection after RPC persistence commits and before NextSteps are queued. Finished, already-canceled, and absent executions are no-ops. RPCs cannot select siblings. The result is mutated and returned for fluent use.

type RecoveryErrorInfo

RecoveryErrorInfo describes the final failure that selected a recovery path. Worker failures preserve their detail and error type. Backend failures use the backend failure type.

A handler reads this value with Context.RecoveryError. It is present in an Execute that follows exhausted WaitFor retries under ProceedOnWaitForFailure. It is also present in a recovery Step selected by StepOptions.ExecuteFailure or FlowTimeoutHandlerOptions.Failure.

Use IsMethodTimeout, IsHeartbeatTimeout, and IsRetryTotalDurationExceeded to recognize the Dex time limits. Do not compare ErrorType with a fixed string for these conditions. Each Dex backend spells these error types differently, and the methods accept every spelling. The methods return false for a nil receiver, so the result of Context.RecoveryError needs no nil check before the call. Other failures keep the error type that the Worker reported, such as an application error type.

package billing

import "github.com/superdurable/dex/sdk-go/dex"

// ChargeRecoveryStep runs after the charge Step exhausts its Execute retries.
type ChargeRecoveryStep struct {
	dex.StepDefaultsNoWaitFor[string]
}

func (ChargeRecoveryStep) Execute(ctx dex.Context, orderID string) (*dex.StepDecision, error) {
	failure := ctx.RecoveryError()
	switch {
	case failure.IsMethodTimeout():
		return dex.ForceFail("charging order " + orderID + " exceeded its method timeout"), nil
	case failure.IsHeartbeatTimeout():
		return dex.ForceFail("charging order " + orderID + " stopped sending heartbeats"), nil
	case failure.IsRetryTotalDurationExceeded():
		return dex.ForceFail("charging order " + orderID + " used its whole retry duration"), nil
	case failure != nil:
		return dex.ForceFail("charging order " + orderID + " failed: " + failure.ErrorType), nil
	default:
		return dex.ForceFail("charging order " + orderID + " has no recorded failure"), nil
	}
}

type RecoveryErrorInfo struct {
    // Detail is the failure message supplied by the Worker or backend.
    Detail string
    // ErrorType is the original Worker error type or backend failure type.
    ErrorType string
}

func (*RecoveryErrorInfo) IsHeartbeatTimeout

func (info *RecoveryErrorInfo) IsHeartbeatTimeout() bool

IsHeartbeatTimeout reports whether the final attempt stopped recording heartbeats in time.

Use it in a recovery path to handle an attempt that did not call Context.RecordHeartbeat within StepOptions.HeartbeatTimeout or FlowTimeoutHandlerOptions.HeartbeatTimeout. See the RecoveryErrorInfo example. A Worker error type that reuses the backend's heartbeat timeout name also matches. Give application error types distinct names.

IsHeartbeatTimeout returns true when ErrorType names a heartbeat timeout, ignoring letter case and underscores. It returns false for every other ErrorType and for a nil receiver.

func (*RecoveryErrorInfo) IsMethodTimeout

func (info *RecoveryErrorInfo) IsMethodTimeout() bool

IsMethodTimeout reports whether the final attempt ran longer than its Step method timeout.

Use it in a recovery path to handle an attempt that exceeded StepOptions.WaitForMethodTimeout, StepOptions.ExecuteMethodTimeout, or FlowTimeoutHandlerOptions.MethodTimeout. See the RecoveryErrorInfo example.

If the call to the Worker fails before Dex records the method timeout, Dex reports a Worker API failure instead, and this method returns false. A Worker API failure has many other causes, such as an unreachable Worker or an invalid response, so do not treat it as a timeout.

A Worker error type that reuses the backend's method timeout name also matches. Give application error types distinct names.

IsMethodTimeout returns true when ErrorType names a method timeout, ignoring letter case and underscores. It returns false for every other ErrorType and for a nil receiver.

func (*RecoveryErrorInfo) IsRetryTotalDurationExceeded

func (info *RecoveryErrorInfo) IsRetryTotalDurationExceeded() bool

IsRetryTotalDurationExceeded reports whether the overall deadline of the Step method call ended the final attempt while it was still running.

Use it in a recovery path to tell this case apart from a method timeout or a heartbeat timeout. See the RecoveryErrorInfo example.

Where the overall deadline comes from depends on the Dex Server:

When the total duration ends between attempts, Dex starts no further attempt. It reports the error type of the final attempt instead, so this method returns false. Check the other methods and ErrorType to handle that case. A Worker error type that reuses the backend's deadline name also matches. Give application error types distinct names.

IsRetryTotalDurationExceeded returns true when ErrorType names the overall deadline, ignoring letter case and underscores. It returns false for every other ErrorType and for a nil receiver.

type Registry

Registry stores immutable Flow definitions shared by Client and Worker.

NewRegistry validates every definition before publishing the Registry. Build one Registry during application startup and reuse it for every Client and Worker that must exchange those Flow types. Registry is safe for concurrent reads after NewRegistry returns.

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

func NewRegistry

func NewRegistry(flows []Flow) (*Registry, error)

NewRegistry validates and assembles Flow definitions atomically.

flows may be empty, but every Flow must be non-nil and have a unique Flow type. The default Flow type is the Go type name without its package, so two Flows from different packages that share a Go type name conflict. NewRegistry also validates starting Steps, Step options, RPC signatures, persistence definitions, and cross-Flow Attribute index compatibility. It returns FlowDefinitionError for an invalid Flow and publishes no partially assembled Registry.

registry, err := dex.NewRegistry([]dex.Flow{OrderFlow{}, RefundFlow{}})
if err != nil {
	return err
}

type RequestTimeoutError

RequestTimeoutError reports that a caller-visible request budget expired.

type RequestTimeoutError struct {
    // ServiceError contains the timed-out wait metadata.
    *ServiceError
}

func (*RequestTimeoutError) Unwrap

func (e *RequestTimeoutError) Unwrap() error

Unwrap returns the shared service failure.

type RetryAfterError

RetryAfterError requests a delay before the next retry while preserving the current failure. Throw the value returned by RetryAfter from waitFor or execute. Dex schedules the next retry after the requested delay while keeping the wrapped failure as the reported Worker error.

type RetryAfterError struct {
    // After is the delay Dex waits before retrying the Step method.
    After time.Duration
    // Cause is the current attempt failure reported to Dex.
    Cause error
}

func RetryAfter

func RetryAfter(after time.Duration, cause error) *RetryAfterError

RetryAfter creates a retry request while preserving the current attempt failure. after must be positive. cause is the Step method failure reported to Dex.

func (*RetryAfterError) Error

func (retryAfter *RetryAfterError) Error() string

Error returns the underlying attempt failure message.

func (*RetryAfterError) Unwrap

func (retryAfter *RetryAfterError) Unwrap() error

Unwrap returns the underlying attempt failure for errors.Is and errors.As.

type RetryPolicy

RetryPolicy overrides retry timing for a Step's WaitFor or Execute method. Zero fields preserve server defaults, including a four-hour total duration. MaximumAttempts counts the initial attempt. With asynchronous durability, local and regular execution share attempts and elapsed duration. Fallback is immediate, while subsequent regular retries continue the cumulative backoff sequence.

type RetryPolicy struct {
    // InitialInterval is the delay before the first retry.
    InitialInterval time.Duration
    // BackoffCoefficient multiplies each successive retry interval.
    BackoffCoefficient float64
    // MaximumInterval caps the delay between retries.
    MaximumInterval time.Duration
    // MaximumAttempts limits total attempts, including the initial attempt.
    MaximumAttempts int32
    // TotalDuration limits elapsed time across all attempts.
    TotalDuration time.Duration
}

type SearchFlowEntry

SearchFlowEntry contains one Flow row returned by SearchFlows.

type SearchFlowEntry struct {
    // FlowID is the application-assigned Flow ID.
    FlowID string
    // RunID is the server-assigned run ID.
    RunID string
    // FlowType is the registered Flow type.
    FlowType string
    // Status is the indexed Flow status.
    Status FlowStatus
    // StartedAt is the indexed start time.
    StartedAt time.Time
    // ClosedAt is zero while open or when the close time is unavailable.
    ClosedAt time.Time
    // IndexedAttributes contains values keyed by their physical index names.
    IndexedAttributes map[string]Value
}

type SearchFlowsOptions

SearchFlowsOptions selects whether visibility searches include earlier Continue-as-New runs.

type SearchFlowsOptions struct {
    // IncludeContinuedAsNew disables the server's default exclusion without changing the caller's query.
    IncludeContinuedAsNew bool
}

type SearchFlowsPage

SearchFlowsPage contains one page of Flow search results.

type SearchFlowsPage struct {
    // Flows are returned in server-defined query order.
    Flows []SearchFlowEntry
    // NextPageToken is opaque and empty on the last page.
    NextPageToken string
}

type ServiceError

ServiceError reports a FlowService failure.

type ServiceError struct {
    // Op is the SDK operation that failed.
    Op  string
    // FlowID is the targeted Flow ID, or empty for service-wide operations.
    FlowID string
    // Code is the outer gRPC status code.
    Code codes.Code
    // SubStatus is Dex's specific error classification.
    SubStatus ErrorSubStatus
    // Detail is the most specific server or transport message available.
    Detail string
    // contains filtered or unexported fields
}

func (*ServiceError) Error

func (e *ServiceError) Error() string

Error formats operation, Flow identity, gRPC code, and detail.

func (*ServiceError) Unwrap

func (e *ServiceError) Unwrap() error

Unwrap returns the original gRPC transport failure.

type StartFlowOptions

StartFlowOptions configures a new Flow execution. Nil and zero fields preserve server defaults; RequestID is generated when omitted.

To retry one logical start safely, for example after a lost response, set a stable RequestID and AlreadyStarted.IgnoreError together. Reuse the same values on every retry:

flowID := "order-" + orderID
requestID := "start-order-" + orderID
runID, err := client.StartFlow(ctx, orderFlow, flowID, input, dex.StartFlowOptions{
	AlreadyStarted: &dex.AlreadyStartedOptions{IgnoreError: true},
	RequestID:      &requestID,
})

type StartFlowOptions struct {
    // Timeout sets Dex's durable soft timeout; nil or zero disables it.
    Timeout *time.Duration
    // TimeoutPolicy controls what Dex does when Timeout expires.
    // The default uses Handler when the Flow implements FlowTimeoutHandler, otherwise Fail.
    TimeoutPolicy FlowTimeoutPolicy
    // TimeoutHandlerOptions configures execution when TimeoutPolicy resolves to TimeoutHandler.
    // It requires a positive Timeout and is invalid with Fail or Cancel.
    TimeoutHandlerOptions *FlowTimeoutHandlerOptions
    // IDReusePolicy controls reuse of an existing Flow ID.
    IDReusePolicy IDReusePolicy
    // StartDelay postpones the first Step after start acceptance.
    StartDelay *time.Duration
    // RetryPolicy configures whole-Flow retries after terminal failures.
    RetryPolicy *FlowRetryPolicy
    // Attributes supplies encoded initial Attribute and Attribute-map values.
    Attributes []InitialAttributeDef
    // ConfigOverride replaces registered Flow configuration for this execution.
    ConfigOverride *FlowConfig
    // AlreadyStarted controls whether an existing Flow is returned as success.
    // IgnoreError returns the existing run only when RequestID matches the request that started it.
    AlreadyStarted *AlreadyStartedOptions
    // RequestID defaults to a UUID; set a stable business identifier for cross-call retries.
    // A retried StartFlow returns the existing run only when AlreadyStarted.IgnoreError is true
    // and RequestID matches the request that started that run.
    // Without IgnoreError, the retry returns FlowAlreadyStartedError even when the first attempt
    // created the Flow.
    // The match applies only when IDReusePolicy rejects the start.
    // A policy that permits another run starts a new run instead.
    // Dex compares only request IDs, not input or other options.
    // Never reuse a request ID for a different logical start.
    // See StartFlowOptions for an example that sets both fields.
    RequestID *string
}

type Step

Step is one node in a Flow's durable state machine. IN is the typed input passed into WaitFor and Execute for this step.

A waiting step embeds StepDefaults and implements WaitFor. An execute-only step embeds StepDefaultsNoWaitFor[IN] (or NoWaitFor[IN]) so registration sets skip_wait_for and never invokes WaitFor.

Example (waiting step):

type ApproveOrderStep struct {
	dex.StepDefaults
}

func (ApproveOrderStep) WaitFor(
	ctx dex.Context,
	input ApproveOrderInput,
) (*dex.Wait, error) {
	return dex.AnyOf(
		ApprovalChannel.ForOne(),
		dex.Timer(input.Timeout, dex.WithConditionID("approval-timeout")),
	), nil
}

func (ApproveOrderStep) Execute(
	ctx dex.Context,
	input ApproveOrderInput,
) (*dex.StepDecision, error) {
	if ctx.HasTimerFired() {
		return dex.ForceFail("approval timed out"), nil
	}
	return dex.GoTo(ShipOrder, ShipOrderInput{OrderID: input.OrderID}), nil
}

var ApproveOrder = ApproveOrderStep{}

Example (execute-only step):

type ShipOrderStep struct {
	dex.StepDefaultsNoWaitFor[ShipOrderInput]
}

func (ShipOrderStep) Execute(
	ctx dex.Context,
	input ShipOrderInput,
) (*dex.StepDecision, error) {
	return dex.DeadEnd(), nil
}

var ShipOrder = ShipOrderStep{}

type Step[IN any] interface {
    // StepSelector supplies this Step's durable type and static options.
    StepSelector

    // WaitFor sets up the conditions this step waits on before Execute runs.
    //
    //	ctx    invocation context (flow id, run id, step execution id, etc.)
    //	input  typed step input for this execution
    //
    // Return a non-nil Wait built from Channel.ForOne / ForAll, Timer, AnyOf / AllOf, or
    // SkipWaitImmediately. Attribute Set, RecordEvent, Publish, and
    // SetStepExecutionLocal are allowed here; writes are accepted with the
    // WaitFor response.
    //
    // Optional for execute-only steps: embed NoWaitFor[IN] or
    // StepDefaultsNoWaitFor[IN] instead of implementing a real WaitFor.
    // Registration detects the marker and skips this method; the panic body must
    // never run.
    WaitFor(ctx Context, input IN) (*Wait, error)

    // Execute decides what happens next after WaitFor conditions complete, or
    // immediately when the step is execute-only (NoWaitFor / StepDefaultsNoWaitFor).
    //
    //	ctx    invocation context; read condition results, timers, and
    //	       step-execution locals set in WaitFor
    //	input  the same typed step input passed to WaitFor
    //
    // Return a non-nil StepDecision (GoTo, GoToMany, DeadEnd, GracefulComplete,
    // ForceComplete, ForceFail, or a conditional close). Attribute / channel /
    // event writes are allowed; they are accepted with the Execute response.
    Execute(ctx Context, input IN) (*StepDecision, error)
}

type StepCompletion

StepCompletion contains one hydrated output from a completed Step.

type StepCompletion struct {
    // StepType is the registered Step type.
    StepType string
    // StepExecutionID is the completed execution's server identity.
    StepExecutionID string
    // Output is an opaque hydrated value decoded with Value.Decode.
    Output Value
}

type StepDecision

StepDecision describes the durable outcome of one successful Step Execute call.

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

func DeadEnd

func DeadEnd() *StepDecision

DeadEnd leaves the Flow running with no next Step. Use it only when an RPC or external operation will resume the Flow later.

func ForceComplete

func ForceComplete(output any) *StepDecision

ForceComplete completes immediately, abandons other active Steps, and returns output.

func ForceCompleteIfChannelsEmpty

func ForceCompleteIfChannelsEmpty(output any, channels []ChannelDef, otherwise ...StepMovement) *StepDecision

ForceCompleteIfChannelsEmpty completes when every guarded Channel is empty. Otherwise Dex follows the movements in otherwise.

func ForceFail

func ForceFail(reason string) *StepDecision

ForceFail ends the Flow immediately with an application-provided reason.

func GoTo

func GoTo[IN any](step Step[IN], input IN, options ...StepMoveOption) *StepDecision

GoTo returns a decision that moves to one typed target Step.

func GoToMany

func GoToMany(movements ...StepMovement) *StepDecision

GoToMany moves to all targets, enabling concurrent active Steps.

func GracefulComplete

func GracefulComplete(output any) *StepDecision

GracefulComplete completes after all other active Steps finish and returns output.

func (*StepDecision) CancelSiblingSteps

func (decision *StepDecision) CancelSiblingSteps(steps ...StepSelector) *StepDecision

CancelSiblingSteps selects executions sharing the current execution's scheduling source.

A sibling has the same Context.FromStepExecutionID value. Snapshot and no-op behavior match CancelSteps. Repeated calls take the union; Flow-wide selection wins for the same Step type.

The decision is mutated and returned for fluent use.

func (*StepDecision) CancelSteps

func (decision *StepDecision) CancelSteps(steps ...StepSelector) *StepDecision

CancelSteps selects every queued or active execution of each registered Step type.

Dex resolves one snapshot after the current Execute succeeds. Finished, already-canceled, and absent executions are no-ops. Steps scheduled by this decision are outside the snapshot. Repeated calls take the union; a Flow-wide selector supersedes a sibling selector.

The decision is mutated and returned for fluent use.

type StepDef

StepDef is the representation of Step, without Go's generic So that internal sdk can use it to workaround Go's generic limitations

StepDef is internal to the SDK. Applications create values with DefineStep or DefineStartStep and return them from Flow.GetSteps.

type StepDef interface {
    // contains filtered or unexported methods
}

func DefineStartStep

func DefineStartStep[IN any](step Step[IN]) StepDef

DefineStartStep wraps the Flow's starting Step for Flow.GetSteps. A Flow may declare at most one starting step. With none, the run starts with no step execution (RPCs can still start steps later).

func DefineStep

func DefineStep[IN any](step Step[IN]) StepDef

DefineStep wraps a non-starting Step for Flow.GetSteps.

type StepDefaults

StepDefaults supplies default type and options while requiring WaitFor.

Example:

type ApproveOrderStep struct {
	dex.StepDefaults
}

type StepDefaults struct {
    // DefaultStepOptions supplies default type naming and nil options.
    DefaultStepOptions
}

type StepDefaultsNoWaitFor

StepDefaultsNoWaitFor supplies defaults for execute-only steps.

Example:

type ShipOrderStep struct {
	dex.StepDefaultsNoWaitFor[ShipOrderInput]
}

type StepDefaultsNoWaitFor[IN any] struct {
    // StepDefaults supplies default type naming and nil options.
    StepDefaults
    // contains filtered or unexported fields
}

type StepDurability

StepDurability controls when staged persistence must be durably acknowledged.

type StepDurability uint8


const (
    // StepDurabilityDefault uses the Flow or server default.
    StepDurabilityDefault StepDurability = iota
    // StepDurabilitySync waits for persistence before completing the handler request.
    StepDurabilitySync
    // StepDurabilityAsync permits persistence after accepting the handler result.
    StepDurabilityAsync
)

type StepExecutionID

StepExecutionID identifies one execution of a Step type within a Flow run.

type StepExecutionID struct {
    // StepType is the stable registered Step type.
    StepType string
    // ExecutionNumber defaults to one when omitted.
    ExecutionNumber *int32
}

type StepMoveOption

StepMoveOption configures one Step movement. Use WithStepOptions to create an option.

type StepMoveOption interface {
    // contains filtered or unexported methods
}

func WithStepOptions

func WithStepOptions(options *StepOptions) StepMoveOption

WithStepOptions overrides the target Step's registered options for one movement.

type StepMovement

StepMovement targets a Step with typed input and optional per-movement StepOptions. Create movements with MovementOf when building multi-target or conditional decisions.

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

func MovementOf

func MovementOf[IN any](step Step[IN], input IN, options ...StepMoveOption) StepMovement

MovementOf creates a typed Step movement for GoToMany, RPCResult, or conditional completion.

type StepOptions

StepOptions configures one Step's handler execution and persistence behavior.

Zero values preserve server defaults. Regular attempts default to two hours with a one-minute heartbeat timeout. Durability resolves from the method override, FlowConfig, then synchronous. Asynchronous durability first uses at most seven local-activity seconds and three attempts. WaitFor failures fail the Flow by default. Each lock is held only for the matching handler call.

type StepOptions struct {
    // WaitForMethodTimeout limits one WaitFor attempt; zero uses the server default.
    WaitForMethodTimeout time.Duration
    // ExecuteMethodTimeout limits one Execute attempt; zero uses the server default.
    ExecuteMethodTimeout time.Duration
    // HeartbeatTimeout detects stalled regular WaitFor and Execute activities.
    // Zero uses the one-minute default. Positive values must be whole seconds within int32 range
    // and meet the server-configured minimum, which defaults to ten seconds.
    HeartbeatTimeout time.Duration
    // WaitForRetry overrides the WaitFor retry policy; nil uses server defaults.
    WaitForRetry *RetryPolicy
    // ExecuteRetry overrides the Execute retry policy; nil uses server defaults.
    ExecuteRetry *RetryPolicy
    // WaitForFailure selects exhausted WaitFor behavior; zero fails the Flow.
    WaitForFailure WaitForFailurePolicy
    // ExecuteFailure selects a recovery Step after Execute exhausts retries.
    ExecuteFailure *ExecuteFailure
    // WaitForDurability overrides persistence durability for WaitFor writes.
    WaitForDurability StepDurability
    // ExecuteDurability overrides persistence durability for Execute writes.
    ExecuteDurability StepDurability
    // WaitForLockAttributes are acquired together for the WaitFor invocation.
    WaitForLockAttributes []AttributeLock
    // ExecuteLockAttributes are acquired together for the Execute invocation.
    ExecuteLockAttributes []AttributeLock
    // WaitForLoadAttributeMaps includes every current instance of each AttributeMap in WaitFor.
    WaitForLoadAttributeMaps []AttributeDef
    // WaitForLoadAttributeMapInstances includes exact AttributeMap instances in WaitFor.
    WaitForLoadAttributeMapInstances []AttributeMapLoad
    // WaitForLoadChannels includes pending messages from selected Channels in WaitFor.
    WaitForLoadChannels []ChannelDef
    // WaitForLoadChannelMaps includes every current ChannelMap instance in WaitFor.
    WaitForLoadChannelMaps []ChannelDef
    // WaitForLoadChannelMapInstances includes exact ChannelMap instance messages in WaitFor.
    WaitForLoadChannelMapInstances []ChannelMapLoad
    // ExecuteLoadAttributeMaps includes every current instance of each AttributeMap in Execute.
    ExecuteLoadAttributeMaps []AttributeDef
    // ExecuteLoadAttributeMapInstances includes exact AttributeMap instances in Execute.
    ExecuteLoadAttributeMapInstances []AttributeMapLoad
    // ExecuteLoadChannels includes pending messages from selected Channels in Execute.
    ExecuteLoadChannels []ChannelDef
    // ExecuteLoadChannelMaps includes every current ChannelMap instance in Execute.
    ExecuteLoadChannelMaps []ChannelDef
    // ExecuteLoadChannelMapInstances includes exact ChannelMap instance messages in Execute.
    ExecuteLoadChannelMapInstances []ChannelMapLoad
}

type StepSelector

StepSelector identifies a registered Step type for cancellation decisions.

Every Step implements StepSelector through GetStepType and GetStepOptions. This non-generic view lets one cancellation request select Steps with different input types.

type StepSelector interface {
    // GetStepType overrides the default Go type name without its package.
    // Embed DefaultStepType to use the default.
    //
    // The Step type is the durable identity of a Step within its Flow. Run
    // history, Dex Web, and StepExecutionID use it as is. The default for
    // *orders.shipOrder is "shipOrder". Step types must be unique within one
    // Flow, and the package does not disambiguate them.
    //
    // Prefer the default name. Return a non-empty compile-time string only to
    // keep an existing durable name after renaming the Go type, or to name a
    // generic Step type, which has no default. Returning an empty string selects
    // the default.
    //
    //	func (shipOrder) GetStepType() string {
    //		return "ShipOrder"
    //	}
    GetStepType() string

    // GetStepOptions returns immutable defaults applied whenever this Step is
    // scheduled. Return nil (e.g. by embedding StepDefaults) to use server
    // defaults. A movement may override individual fields when transitioning.
    GetStepOptions() *StepOptions
}

type StopOptions

StopOptions configures Client.StopFlow.

type StopOptions struct {
    // Type selects cancel, terminate, or fail behavior.
    Type StopType
    // Reason is recorded with the stop operation.
    Reason string
}

type StopType

StopType selects how StopFlow ends an active Flow.

type StopType uint8


const (
    // CancelFlow requests cooperative cancellation.
    CancelFlow StopType = iota + 1
    // TerminateFlow ends the Flow immediately without cleanup.
    TerminateFlow
    // FailFlow marks the Flow failed with Reason.
    FailFlow
)

type Stream

Stream defines one typed, best-effort resumable message stream.

Register the Stream in exactly one Flow's PersistenceSchema. All Flow instances sharing that Flow type and Stream name share streamCapacityBytes. Stream writes are immediately visible and do not roll back when a Step later fails.

var Thinking = dex.DefineStream[string]("thinking", 10<<20)

func (ThinkStep) Execute(ctx dex.Context, input Input) (*dex.StepDecision, error) {
	if err := Thinking.Write(ctx, "checking inventory"); err != nil {
		return nil, err
	}
	return dex.GracefulComplete(), nil
}

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

func DefineStream

func DefineStream[T any](name string, streamCapacityBytes int64) Stream[T]

DefineStream creates a typed Stream with a stable name and shared approximate byte budget.

streamCapacityBytes must be positive. Validation occurs when NewRegistry registers the Stream.

func (Stream[T]) StreamCapacityBytes

func (s Stream[T]) StreamCapacityBytes() int64

StreamCapacityBytes returns the approximate shared byte budget.

func (Stream[T]) StreamName

func (s Stream[T]) StreamName() string

StreamName returns the stable name sent to Dex.

func (Stream[T]) Write

func (s Stream[T]) Write(ctx Context, value T) error

Write appends value immediately from the current Step invocation.

A Step execution may call Write any number of times. Each call sends a fire-and-forget frame on the current Worker response stream. The returned error covers local validation, encoding, and gRPC transport only; Stream Store failures are observable through server logs and metrics. RPC contexts are rejected. Write and Context.RecordHeartbeat may run concurrently.

type StreamDef

StreamDef is the sealed schema-erasure interface implemented by Stream.

Applications create Stream values with DefineStream and add them to PersistenceSchema.Streams.

type StreamDef interface {
    // contains filtered or unexported methods
}

type StreamMessage

StreamMessage describes one retained Stream message returned by Client.ReadStream.

ResumeToken identifies this message and can be passed unchanged to the next ReadStream call. CreatedTime is assigned by the Stream Store. Source identifies where the message originated.

type StreamMessage struct {
    // ResumeToken identifies this message for the next ReadStream call.
    ResumeToken string
    // CreatedTime is the server-assigned creation time.
    CreatedTime time.Time
    // Source is client-supplied metadata or the Step's #stepExecutionID value.
    Source string
}

type StreamMessagesPage

StreamMessagesPage contains one newest-first page returned by Client.ListStreamMessages.

Pass NextPageToken unchanged as beforePageToken to read the next, older page.

type StreamMessagesPage[T any] struct {
    // Messages contains retained Stream messages in newest-first order.
    Messages []ListedStreamMessage[T]
    // NextPageToken is opaque and empty on the last page.
    NextPageToken string
}

type SubFlowOptions

SubFlowOptions configures one durable SubFlow Condition.

Zero values inherit normal Flow start defaults and use RestartSubFlowIfPreviousExitedAbnormally. The server owns the generated Flow ID and request ID; callers cannot override either value. Attributes must belong to the target SubFlow.

type SubFlowOptions struct {
    // Timeout limits total SubFlow execution duration.
    Timeout *time.Duration
    // TimeoutPolicy controls what Dex does when Timeout expires.
    // The default uses Handler when the SubFlow implements FlowTimeoutHandler, otherwise Fail.
    TimeoutPolicy FlowTimeoutPolicy
    // TimeoutHandlerOptions configures execution when TimeoutPolicy resolves to TimeoutHandler.
    // It requires a positive Timeout and is invalid with Fail or Cancel.
    TimeoutHandlerOptions *FlowTimeoutHandlerOptions
    // StartDelay postpones the SubFlow starting Step after start acceptance.
    StartDelay *time.Duration
    // RetryPolicy configures whole-Flow retries after terminal failures.
    RetryPolicy *FlowRetryPolicy
    // Attributes supplies encoded initial Attribute and Attribute-map values.
    Attributes []InitialAttributeDef
    // ConfigOverride overrides fields inherited from the parent Flow configuration.
    ConfigOverride *FlowConfig
    // ReusePolicy controls how an execution already using the generated Flow ID is resolved.
    ReusePolicy SubFlowReusePolicy
    // ConditionID assigns the stable ID required by AnyComboOf.
    ConditionID string
}

type SubFlowReusePolicy

SubFlowReusePolicy controls how a generated SubFlow Flow ID resolves an existing execution.

type SubFlowReusePolicy uint8


const (
    // RestartSubFlowIfPreviousExitedAbnormally attaches to running executions, returns completed
    // results, and replaces failed, canceled, timed-out, or terminated executions.
    RestartSubFlowIfPreviousExitedAbnormally SubFlowReusePolicy = iota
    // AttachSubFlow attaches to a running execution or returns its existing terminal result.
    AttachSubFlow
    // AlwaysRestartSubFlow replaces a different existing execution, including a running one.
    AlwaysRestartSubFlow
)

type TimeTravelOptions

TimeTravelOptions configures Client.TimeTravel with exactly one historical point.

type TimeTravelOptions struct {
    // Type selects which time travel point field Dex reads.
    Type TimeTravelType
    // Reason is recorded with the time travel operation; empty leaves it unspecified.
    Reason string
    // HistoryEventTime supplies the time for TimeTravelByHistoryEventTime.
    HistoryEventTime time.Time
    // StepType supplies the stable Step type for TimeTravelByStepType.
    StepType string
    // StepExecutionID supplies the exact execution for TimeTravelByStepExecutionID.
    StepExecutionID string
    // StepMethod selects WaitFor or Execute when Type is TimeTravelByStepExecutionID.
    StepMethod TimeTravelStepMethod
    // SkipWritesReapply prevents replay of later RPCs, Channel publications, and Attribute writes; false reapplies them.
    SkipWritesReapply bool
}

type TimeTravelStepMethod

TimeTravelStepMethod selects the Step method used as a time travel boundary.

type TimeTravelStepMethod uint8


const (
    // TimeTravelStepWaitFor resumes before the selected Step execution's WaitFor method.
    TimeTravelStepWaitFor TimeTravelStepMethod = iota + 1
    // TimeTravelStepExecute resumes before the selected Step execution's Execute method.
    TimeTravelStepExecute
)

type TimeTravelType

TimeTravelType identifies the historical point selector used by Client.TimeTravel.

type TimeTravelType uint8


const (
    // TimeTravelToBeginning resumes before the first history event.
    TimeTravelToBeginning TimeTravelType = iota + 1
    // TimeTravelByHistoryEventTime resumes at the last eligible event by time.
    TimeTravelByHistoryEventTime
    // TimeTravelByStepType resumes before the first execution of a Step type.
    TimeTravelByStepType
    // TimeTravelByStepExecutionID resumes before an exact Step execution.
    TimeTravelByStepExecutionID
)

type TimerID

TimerID selects one Timer Condition within a Step execution.

type TimerID struct {
    // ConditionID targets a Condition configured with WithConditionID.
    ConditionID string
    // Index targets a zero-based Timer Condition position when non-nil.
    Index *int32
}

type Value

Value is an opaque Dex value. Decoded string values contain valid UTF-8.

Values returned by Client metadata reads are already hydrated. Use Decode with the same Go type used by the matching Attribute. Primitive values preserve their wire kind; structs, maps, and non-byte slices use JSON; byte slices use the raw-bytes encoding.

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

func (Value) Decode

func (value Value) Decode(valuePtr any) error

Decode decodes Value into a non-nil pointer.

valuePtr must be a non-nil pointer compatible with the encoded wire value. Decode replaces the pointed-to value on success and returns a ValueMappingError for an incompatible target, malformed payload, unsupported encoding, or numeric overflow.

page, err := client.SearchFlows(ctx, "CustomKeywordField = 'ready'", 1, "")
if err != nil {
	return err
}
var status string
err = page.Flows[0].IndexedAttributes["CustomKeywordField"].Decode(&status)

type ValueMappingError

ValueMappingError reports a Dex value conversion failure.

type ValueMappingError struct {
    // Operation describes whether encoding, decoding, indexing, or hydration failed.
    Operation string
    // Err preserves the serializer or type-conversion failure.
    Err error
}

func (*ValueMappingError) Error

func (e *ValueMappingError) Error() string

Error formats the value operation and underlying mapping failure.

func (*ValueMappingError) Unwrap

func (e *ValueMappingError) Unwrap() error

Unwrap returns the underlying value-mapping failure.

type Wait

Wait describes when Dex may invoke a Step's Execute method.

Build waits from durable Timer and Channel Conditions. Condition order determines timer indexes used by TimerID and HasTimerFiredByIndex.

Example:

return dex.AnyOf(
	approvals.ForOne(),
	dex.Timer(5*time.Minute),
), nil

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

func AllOf

func AllOf(conditions ...Condition) *Wait

AllOf waits until every condition is satisfied. Conditions do not need IDs.

func AnyComboOf

func AnyComboOf(combinations ...ConditionCombination) *Wait

AnyComboOf waits until every Condition in at least one combination is satisfied. It consumes messages only from Channel conditions in the selected combination. Every Condition requires a non-empty user-provided ID.

func AnyOf

func AnyOf(conditions ...Condition) *Wait

AnyOf waits until at least one condition is satisfied. Conditions do not need IDs. Among ready conditions, Dex selects the first in canonical order: Timers, Channels, then SubFlows, preserving declaration order within each kind. An unready condition does not block a later ready condition. Return only the higher-priority condition to wait for strict priority. It consumes messages only from the selected Channel condition, not other ready alternatives.

func SkipWaitImmediately

func SkipWaitImmediately() *Wait

SkipWaitImmediately asks Dex to invoke Execute without evaluating WaitFor conditions.

func Until

func Until(condition Condition) *Wait

Until waits until one condition is satisfied. The Condition does not need an ID.

type WaitForAttributeOptions

WaitForAttributeOptions configures one durable Attribute match wait.

When RequestID is empty, the server derives a stable ID from the Attribute condition. Reuse an override only when retrying the same Attribute match. RequestTimeout bounds the caller-visible request across transparent transport reattachments. Leave it zero to wait indefinitely. InternalHandlerTimeout bounds how long accepted waits left behind by caller timeout or cancellation consume Temporal's 10 in-flight Update slots per Workflow Execution. Active callers transparently start a new generation. Leave it zero unless abandoned waits can approach that limit. Each rollover adds another Update to history, which has a 2,000-Update limit. Both durations must use whole seconds within int32 range.

options := dex.WaitForAttributeOptions{
	RequestTimeout: time.Second * 10,
}

type WaitForAttributeOptions struct {
    // RequestID overrides the stable ID derived from the Attribute condition.
    RequestID string
    // RequestTimeout bounds the caller-visible request. Zero waits indefinitely.
    RequestTimeout time.Duration
    // InternalHandlerTimeout bounds one internal Temporal Update handler generation.
    // Set it only to reclaim abandoned waits from Temporal's in-flight Update limit.
    // Zero disables rollover. Each rollover adds another Update to history.
    InternalHandlerTimeout time.Duration
}

type WaitForFailurePolicy

WaitForFailurePolicy selects behavior after WaitFor exhausts its retry policy.

type WaitForFailurePolicy uint8


const (
    // FailFlowOnWaitForFailure ends the Flow after WaitFor exhausts retries.
    FailFlowOnWaitForFailure WaitForFailurePolicy = iota
    // ProceedOnWaitForFailure invokes Execute and exposes the failure through Context.
    ProceedOnWaitForFailure
)

type WaitForFlowOptions

WaitForFlowOptions controls Flow-result hydration.

type WaitForFlowOptions struct {
    // NeedsResults asks Dex to include completed Step outputs.
    NeedsResults bool
}

type WaitForStepCompletionOptions

WaitForStepCompletionOptions configures one durable Step completion wait.

When RequestID is empty, the server derives a stable ID from the Step execution. Reuse an override only when retrying the same Step completion wait. RequestTimeout bounds the caller-visible request across transparent transport reattachments. Leave it zero to wait indefinitely. InternalHandlerTimeout bounds how long accepted waits left behind by caller timeout or cancellation consume Temporal's 10 in-flight Update slots per Workflow Execution. Active callers transparently start a new generation. Leave it zero unless abandoned waits can approach that limit. Each rollover adds another Update to history, which has a 2,000-Update limit. Both durations must use whole seconds within int32 range.

options := dex.WaitForStepCompletionOptions{
	RequestTimeout: time.Second * 10,
}

type WaitForStepCompletionOptions struct {
    // RequestID overrides the stable ID derived from the Step execution.
    RequestID string
    // RequestTimeout bounds the caller-visible request. Zero waits indefinitely.
    RequestTimeout time.Duration
    // InternalHandlerTimeout bounds one internal Temporal Update handler generation.
    // Set it only to reclaim abandoned waits from Temporal's in-flight Update limit.
    // Zero disables rollover. Each rollover adds another Update to history.
    InternalHandlerTimeout time.Duration
}

type Worker

Worker hosts registered Flows through outbound pull streams.

A Worker is one-shot: construct it, call Start once, then call Stop during shutdown. Start blocks while serving. Step and RPC handlers may execute concurrently, so application handler state must provide its own synchronization.

worker, err := dex.NewWorker(registry, cache, dex.WorkerOptions{})
if err != nil {
	return err
}
go func() { serveErr <- worker.Start() }()
defer worker.Stop(context.Background())

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

func NewWorker

func NewWorker(registry *Registry, cache *blobcache.Cache, options WorkerOptions) (*Worker, error)

NewWorker constructs a one-shot Worker from shared dependencies.

registry supplies definitions and cache hydrates large values. It panics when registry or cache is nil. It validates the FlowService address, FlowServiceTLS, synchronization timeout, GrpcMaxMessageBytes, and pull options, then creates an outbound FlowService connection without opening task streams. The returned Worker owns that connection; Start or Stop closes it.

func (*Worker) PullRegistrationState

func (worker *Worker) PullRegistrationState() PullRegistrationState

PullRegistrationState returns an independent snapshot of all pull queue registrations.

Use IsReady for a readiness probe and Queues for diagnostics. It is safe to call concurrently before, during, or after Start and Stop. The snapshot does not flap for reconnects or transient errors after initial registration; only a permanent rejection on a queue with no registered stream makes it unready. Re-registration clears that queue's error. Stop alone does not clear latched readiness, so remove a stopping Worker from service separately. See WaitPullRegistered for a complete HTTP probe and startup gating example.

func (*Worker) PullReservedRPCDeclines

func (worker *Worker) PullReservedRPCDeclines() uint64

PullReservedRPCDeclines returns the cumulative number of Step tasks declined for reserved RPC capacity. It counts decline events; Dex may decline the same Step dispatch repeatedly while retrying it. Read its rate to assess reserve pressure. It excludes full-capacity rejections and unknown names. It never resets during this Worker's lifetime. Calls are safe alongside Start, Stop, and handlers. Export the value to a monitoring counter to distinguish reserve pressure from full capacity:

declines := worker.PullReservedRPCDeclines()
log.Printf("reserved RPC capacity declines: %d", declines)

func (*Worker) Start

func (worker *Worker) Start() (startErr error)

Start serves Worker tasks and blocks until Stop or a serve failure.

Start may be called exactly once. Before serving, it negotiates a Server protocol, synchronizes Attribute indexes, and waits up to WorkerOptions.AttributeIndexSyncTimeout for each startup request. It returns nil after a normal Stop, or an error for synchronization, polling, or invalid lifecycle state failures. Call Start on a dedicated goroutine when the application must continue doing other work. Poll stream failures are retried while the Worker runs; registration failures are logged with a longer retry delay unless PullWorkerOptions.StopOnPermanentStreamRejection is enabled.

func (*Worker) Stop

func (worker *Worker) Stop(ctx context.Context) error

Stop stops accepting tasks and drains handlers until ctx or StopDrainTimeout expires.

Stop accepts calls before Start and repeated calls after shutdown. A nil context is invalid. When the drain deadline expires, Stop requests cancellation, waits for handlers and connection cleanup, and returns the context error. A handler ignoring cancellation can block Stop. PullWorkerOptions.StopDrainTimeout defaults to 30 seconds. Stop closes the Worker's internal FlowService connection before returning.

func (*Worker) WaitPullRegistered

func (worker *Worker) WaitPullRegistered(ctx context.Context) error

WaitPullRegistered waits until every pull queue has received a Registered acknowledgement at least once.

Use it to gate application startup and to start replacement Workers before stopping old Workers during a rolling deployment. Call Start in another goroutine. WaitPullRegistered is safe before or after Start and from multiple goroutines. Once all queues have registered, every later call returns nil, including after Stop; use PullRegistrationState for subsequent readiness probes. A successful wait establishes registration, not available handler capacity.

Before initial readiness, it returns the first *PullRegistrationError immediately on a permanent rejection when that queue has no other registered stream, or Start's error if startup fails. Surplus stream rejections are recorded for diagnostics and retried while another stream holds the queue. Wrapped startup errors retain errors.Is and errors.As behavior. Stop before readiness returns an error. A nil ctx returns an argument error. Otherwise ctx bounds only this wait; cancellation returns ctx.Err without stopping the Worker. A completed registration or recorded failure takes precedence over cancellation.

func servePullWorker(ctx context.Context, worker *dex.Worker, mux *http.ServeMux) error {
	result := make(chan error, 1)
	go func() { result <- worker.Start() }()
	startupCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
	registrationErr := worker.WaitPullRegistered(startupCtx)
	cancel()
	if registrationErr != nil {
		return errors.Join(registrationErr, worker.Stop(context.Background()), <-result)
	}
	mux.HandleFunc("GET /readyz", func(w http.ResponseWriter, r *http.Request) {
		if !worker.PullRegistrationState().IsReady {
			http.Error(w, "pull queues unavailable", http.StatusServiceUnavailable)
			return
		}
		w.WriteHeader(http.StatusOK)
	})
	select {
	case err := <-result:
		return errors.Join(err, worker.Stop(context.Background()))
	case <-ctx.Done():
		return errors.Join(worker.Stop(context.Background()), <-result)
	}
}

type WorkerError

WorkerError preserves the original Worker task failure.

type WorkerError struct {
    // Code is the original Worker gRPC status.
    Code codes.Code
    // Type is the original Worker error or exception type.
    Type string
    // Detail is the original Worker-provided message.
    Detail string
}

type WorkerInvocationError

WorkerInvocationError reports a Worker task or user-handler failure.

type WorkerInvocationError struct {
    // ServiceError contains the outer Dex service failure.
    *ServiceError
    // Worker preserves the original Worker-side status, type, and detail.
    Worker *WorkerError
}

func (*WorkerInvocationError) Unwrap

func (e *WorkerInvocationError) Unwrap() error

Unwrap returns the outer service failure.

type WorkerOptions

WorkerOptions configures a Worker that receives tasks over outbound streams.

Zero values connect to the local Dex Server with plaintext and default pull settings. Workers need only an outbound connection to the Dex Server.

type WorkerOptions struct {
    // Pull configures task capacity, registration, and shutdown. Nil uses PullWorkerOptions defaults.
    Pull *PullWorkerOptions
    // FlowServiceAddress is the Dex gRPC endpoint used for startup synchronization, task streams, and blob hydration. Default: "localhost:8801".
    // With FlowServiceTLS, use host:port, dns:///host:port, or dns://authority/host:port, such as "dex.example.com:443".
    // Startup synchronization registers Attribute indexes.
    FlowServiceAddress string
    // FlowServiceTLS enables TLS for this Worker's FlowService connection. Nil uses plaintext.
    // See FlowServiceTLSOptions for trust roots, server name, and a complete example.
    FlowServiceTLS *FlowServiceTLSOptions
    // AttributeIndexSyncTimeout bounds index registration at startup. Default: two minutes.
    // Negative durations are invalid.
    AttributeIndexSyncTimeout time.Duration
    // Logger defaults to the shared BlobCache logger.
    // It receives concurrent structured Worker logs and falls back to slog.Default.
    Logger Logger
    // Credential is sent as a bearer credential with the Worker's FlowService calls: startup index
    // verification, task streams, and blob hydration. Default: nil sends none. A Dex server with authentication enabled requires it.
    Credential Credential
    // GrpcMaxMessageBytes limits the gRPC messages the Worker receives, in bytes: Step and RPC
    // invocations on outbound task streams and FlowService responses such as each LoadBlobs
    // response used to hydrate values. Zero uses DefaultGrpcMaxMessageBytes (32 MiB) and keeps
    // gRPC's unlimited send limit. A positive value also limits the messages the Worker sends,
    // including Step and RPC results. Negative values are invalid. The Dex server applies its own
    // api.grpcMaxMessageBytes, and the smaller limit wins.
    // Keep it above the largest application value plus headroom for the message envelope.
    GrpcMaxMessageBytes int
}

Generated by gomarkdoc