gush

package module
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Nov 5, 2026 License: MIT Imports: 39 Imported by: 0

README

Gush

CI Go Reference

A small Go layer over Temporal: write durable flows and steps as plain functions.

Gush needs a Temporal server and does not replace one. Temporal provides the durability, retries and scaling. Gush puts a smaller surface on top:

  • One signature. Flows and steps are func(ctx) (any, error) and decode their input when they need it.
  • Logs that follow the work. Attributes set on a run reach its child flows and steps on any worker, with export over OTLP, so one flow_id finds every line of a flow and its steps.
  • Scheduling by name. Priority and fair scheduling read as options, and children inherit them.
  • Compressed payloads. Inputs and outputs are compressed with zstd, which leaves more room under Temporal's size limits.
  • Readable errors. A failure reads [Declined] card declined rather than Temporal's wrapped chain, and keeps its type and retryability.

Two trade-offs come with that:

  • Inputs and outputs are any, so the compiler does not check that a flow and its step agree on a type.
  • Temporal sees compressed bytes, so its Web UI and CLI cannot show inputs and outputs as text.

The SDK stays within reach: ctx.Context in a flow is the SDK's own workflow.Context, for functions like workflow.Sleep.

Install

Requires Go 1.27. Build with its latest patch release, which has the security fixes.

go get github.com/batchstream/gush

Quick start

Start a local Temporal server, with its CLI:

temporal server start-dev

Then define a step, a flow and a worker. This is examples/quickstart, trimmed:

// Charge is a step: real work, retried on failure.
func Charge(ctx *gush.StepContext) (any, error) {
    var order Order
    err := ctx.Decode(&order)
    if err != nil {
        return nil, err
    }

    ctx.Printf("charging order %s", order.ID)
    receipt := Receipt{OrderID: order.ID, Charged: order.Amount}
    return receipt, nil
}

// Fulfill is a flow: deterministic orchestration of steps.
func Fulfill(ctx *gush.FlowContext) (any, error) {
    var receipt Receipt
    err := ctx.RunStep(Charge, ctx.Input, &receipt, gush.WithMaxAttempts(3))
    return receipt, err
}

func main() {
    var options gush.Options
    client, err := gush.Dial(options)
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    worker := gush.NewWorker("orders").WithSteps(Charge).WithFlows(Fulfill)
    go client.Serve(worker)

    order := Order{ID: "order-1", Amount: 42}
    var receipt Receipt
    request := gush.FlowRequest{Queue: "orders", Flow: Fulfill, Input: order, Output: &receipt}
    err = client.RunFlow(context.Background(), request, gush.WithTimeout(time.Minute))
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("charged %d for %s\n", receipt.Charged, receipt.OrderID)
}

Run it with go run ./examples/quickstart.

Temporal replays flow code to restore it, so flows must be deterministic: no I/O, no time.Now and no time.Sleep. Do real work in steps.

Running flows

RunFlow waits for the result. StartFlow returns the flow's ID once it has started.

request := gush.FlowRequest{Queue: "orders", Flow: Fulfill, Input: order}
flowID, err := client.StartFlow(ctx, request)

flow, err := client.GetFlow(ctx, flowID) // flow.Status: Running, Completed, Failed, ...
err = client.CancelFlow(ctx, flowID)     // lets the flow clean up
err = client.TerminateFlow(ctx, flowID, "reason") // stops it immediately

A flow starts a child the same way. A child with no queue runs on its parent's. One begun with StartFlow is independent: it keeps running when its parent ends or is canceled, and CancelFlow stops it by the returned ID.

request := gush.FlowRequest{Flow: Notify, Input: receipt}
err := ctx.RunFlow(request)

Flows and steps are registered under their bare Go function names, and can be called by that name as a string. Two functions with the same name in different packages collide, and Serve reports it. Use NewFlowHandler and NewStepHandler to choose a name, which closures and the instantiations of a generic function need too.

Strings and byte slices, through any number of pointers and including types built on them, travel as raw bytes unless the type has its own JSON or text methods. Everything else travels as JSON, so read a value back as the type it was written. A pointer output stays nil when the output is empty. Adding or removing those methods changes how such a type encodes, so let flows carrying it finish first, and do the same when release notes say an upgrade changes how values travel.

Parallel steps and cleanup

RunStep and RunFlow run on the context their FlowContext holds, and StartFlow only checks that it is not canceled. Code on another workflow.Context gets a matching FlowContext from WithContext, as a goroutine does:

group := workflow.NewWaitGroup(ctx.Context)
for _, item := range items {
    group.Add(1)
    workflow.Go(ctx.Context, func(g workflow.Context) {
        defer group.Done()
        err := ctx.WithContext(g).RunStep(Process, item, nil)
        if err != nil {
            ctx.Printf("process %s: %v", item, err)
        }
    })
}
group.Wait(ctx.Context)

After CancelFlow, anything on the flow's own context returns a canceled error. Cleanup needs a context that ignores the cancellation, and WithWaitForCancel so it starts after the step has ended rather than alongside it:

err := ctx.RunStep(Charge, order, &receipt, gush.WithWaitForCancel())
if temporal.IsCanceledError(err) {
    disconnected, _ := workflow.NewDisconnectedContext(ctx.Context)
    return nil, ctx.WithContext(disconnected).RunStep(Refund, order, nil)
}

A step that is already running only finds out about the cancellation through ctx.Heartbeat(), and WithHeartbeatTimeout makes that prompt. It should then return on ctx.Done().

Options
Option Applies to Effect
WithMaxAttempts(n) both Total attempts, including the first. Zero means the default, and gush.UnlimitedAttempts keeps retrying. Any other negative number is an error.
WithInitialInterval(d) both Delay before the first retry.
WithBackoffCoefficient(f) both Multiplier on the delay after each retry.
WithMaxInterval(d) both Cap on the delay between retries.
WithRetryPolicy(p) both Replaces all retry settings.
WithNoRetry(types...) both Error types that fail immediately.
WithNoRetryErrors(errs...) both The same, taken from *ApplicationError values.
WithPriority(p) both Scheduling priority.
WithFairKey(key) both Group to schedule fairly against, such as a tenant.
WithLogAttribute(k, v) both Attribute on every log line of this run and its children.
WithLogAttributes(m) both The same, for several attributes.
WithID(id) flow Flow ID. Otherwise one is generated, which StartFlow returns. A client's RunFlow or StartFlow joins a flow already running with this ID and ignores this run's flow, input and settings, so concurrent callers share one run. A child flow gets an error that gush.IsFlowRunningError recognizes.
WithErrorIfRunning() flow A client's RunFlow or StartFlow returns that same error instead of joining.
WithTimeout(d) flow Limit on the whole execution, including every retry. A negative one is an error.
WithConfig(c) flow Starts from a FlowConfig, such as one decoded from a queue. Settings left out of the JSON, or zero, use their defaults, and durations in JSON are integer nanoseconds. WithErrorIfRunning stays on once set. Whoever applies it chooses RunFlow or StartFlow.
WithQueue(q) step Runs on another queue.
WithAttemptTimeout(d) step Limit on each attempt, not the step as a whole. A negative one is an error, and so is zero, unless WithPickupTimeout and a limited number of attempts give the step an overall limit.
WithHeartbeatTimeout(d) step Fails an attempt that stops calling ctx.Heartbeat(). A negative one is an error.
WithPickupTimeout(d) step Fails if no worker picks the step up in time.
WithWaitForCancel() step When the flow is canceled, waits for the running step to end instead of returning at once.

Retries and errors

Flow Step
Attempts 1 10
Timeout none 2 minutes per attempt
Backoff 1s, doubling, up to 100s 1s, doubling, up to 100s

Any error returned from a step or flow is retried until attempts run out, including one from errors.New. There are three ways to stop that:

// Fail this step now, whatever the retry policy says.
return nil, ctx.Abort(err)

// The same, with a type the caller can match.
return nil, gush.NewPermanentError("card declined", "Declined", err)

// Let the caller decide, per run.
err := ctx.RunStep(Charge, order, &receipt, gush.WithNoRetry("Declined"))

WithNoRetry matches the type of a gush.NewApplicationError. Pass gush.StepTimeoutErrorType to stop retrying steps that timed out. A TemporalTimeout: entry naming a timeout Temporal does not know is an error when there can be a retry.

Errors from RunFlow and RunStep work with errors.As, gush.IsApplicationError, gush.IsPermanentError and gush.IsPanicError. gush.IsInvalidConfigError recognizes a request refused as invalid, such as settings Temporal refuses or an input over its size limit, so sending it again cannot help. Wrapping one with fmt.Errorf and %w keeps its type and retryability. Its details stay on the original error, so errors.As can find an outer error without them. Call errors.Unwrap on that one until Details succeeds.

Priority and fairness

err := client.RunFlow(ctx, request,
    gush.WithPriority(gush.PriorityHigh),
    gush.WithFairKey("tenant-123"),
)

Priorities run from PriorityUrgent, PriorityHigh, PriorityMedium (the default) and PriorityLow to PriorityTrivial. A fair key stops one tenant from starving the others on a queue. Temporal rejects keys over 64 bytes, so Gush cuts longer ones to fit, between characters. Keys that agree up to the cut share a group. Child flows and steps inherit both settings.

Both depend on the server, as Temporal's documentation describes. Priority is on by default from Temporal Server 1.31; on 1.29 and 1.30, set matching.useNewMatcher=true. Fairness needs matching.enableFairness=true on a self-hosted server. make local starts a dev server with both.

Suspend and resume

A step can finish later, when a person or another system decides the result. Suspend frees the worker in the meantime.

func ApproveExpense(ctx *gush.StepContext) (any, error) {
    err := sendApprovalRequest(ctx, ctx.ResumeToken, ctx.Input)
    if err != nil {
        return nil, err
    }
    return ctx.Suspend()
}

// In the flow. The attempt timeout keeps running while the step waits.
var decision Decision
err := ctx.RunStep(ApproveExpense, expense, &decision,
    gush.WithAttemptTimeout(24*time.Hour),
    gush.WithMaxAttempts(1),
)

// Anywhere else, once decided.
err = client.Resume(ctx, resumeToken, decision, nil)

Set WithAttemptTimeout to the longest acceptable wait. The step's timeout keeps running while it is suspended, and when it expires the step retries and runs its code again, sending a second request. The default is 2 minutes and 10 attempts. Use WithMaxAttempts(1) unless running the code twice is safe.

An error passed to Resume fails the step for good, so a rejection does not send the request again. To run the code again under the step's retry policy, pass a gush.NewApplicationError. A canceled error cancels the step if it is being canceled, and otherwise fails it for good too.

A token belongs to one attempt. gush.IsStaleResumeTokenError reports that the attempt is over: the step ended, or it is retrying under a new token. The server cannot tell this from a well-formed token that matches no step, such as a corrupted one. gush.IsResumeTokenError also covers a token the server cannot read. If a flow has moved on from a canceled step that has not yet ended on the server, Resume succeeds and the result is discarded.

Changing running flows

Flows can run for a long time, and a replay must take the same path it took originally. Version keeps old executions on the old code.

v := ctx.Version("payment-provider", gush.DefaultFlowVersion, 1)
if v == gush.DefaultFlowVersion {
    err = ctx.RunStep(OldCharge, input, &output)
} else {
    err = ctx.RunStep(NewCharge, input, &output)
}

The second argument is the oldest version the code still supports. Once no execution can reach an old branch, delete it and raise that minimum, here with ctx.Version("payment-provider", 1, 1). An execution that recorded something older then stops making progress, with its flow tasks failing, until code that supports it is deployed. It never runs a branch it did not take.

Long-running flows

A flow's history grows with every step, and Temporal limits its size. A flow that loops for a long time ends the run and starts again with an empty history, keeping its flow ID and log attributes:

if batch.Done {
    return summary, nil
}
return nil, ctx.ContinueAsNew(batch.Next)

Logging

Every flow and step context logs directly, and so does a gush.NewLogger() anywhere else.

func ProcessData(ctx *gush.StepContext) (any, error) {
    ctx.Time("processing")
    ctx.Printf("processing %d bytes", len(ctx.Input))
    ctx.PrintWithAttributes("batch ready", gush.Attr("batch_size", 10))
    ctx.TimeEnd("processing", "done")
    return nil, nil
}

Each record carries flow_id, the source file and line, and the attributes inherited from the run. A child flow has its own flow_id, so set an attribute where a run starts to tie the whole tree together:

err := client.RunFlow(ctx, request, gush.WithLogAttribute("tenant", "acme"))

Lines written while Temporal replays a flow are dropped, so each appears once. Helpers that only have a context.Context, even one derived with context.WithTimeout, can log with the step's context through gush.AsStepContext(ctx).Printf(...).

Export over OTLP

gush.Dial starts exporting when OTEL_EXPORTER_OTLP_LOGS_ENDPOINT or OTEL_EXPORTER_OTLP_ENDPOINT is set, and Close flushes. A process that logs without a client calls gush.InitLogExport() itself, and one that builds its own LoggerProvider gets the same exporter from gush.NewLogExporter(ctx). OpenTelemetry's other variables set the headers and TLS. The protocol is http/protobuf unless they ask for grpc, and any other protocol is an error:

OTEL_EXPORTER_OTLP_ENDPOINT=https://collector:4317
OTEL_EXPORTER_OTLP_PROTOCOL=grpc

The endpoint must be a URL, or Dial returns an error. https:// is TLS and http:// is plaintext, and OTEL_EXPORTER_OTLP_INSECURE does not change that. Give the port too, because leaving it out does not select OTLP's 4317 or 4318. Over HTTP, /v1/logs is added to OTEL_EXPORTER_OTLP_ENDPOINT, but OTEL_EXPORTER_OTLP_LOGS_ENDPOINT is used as written, so it needs the full path, such as http://collector:4318/v1/logs.

An application with its own logging pipeline keeps control: a provider set with OpenTelemetry's global.SetLoggerProvider always wins, and Gush uses it instead of starting an export. Importing Gush never changes that global. With no provider and no endpoint, logs go to standard error, as do the Temporal SDK's own messages unless Options.Logger says otherwise.

Workers

worker := gush.NewWorker("orders").
    WithSteps(Charge, Ship).
    WithFlows(Fulfill).
    WithMaxSteps(100).
    WithStepRate(50).
    WithStopTimeout(time.Minute)

err := client.Serve(worker)

Serve blocks until the process receives SIGINT or SIGTERM or the worker's WithContext is canceled. Run more workers on other machines to share a queue. GetQueueStats reports backlog and rates for scaling decisions:

stats, err := client.GetQueueStats(ctx, "orders")
if err != nil {
    return err
}
fmt.Println(stats[gush.QueueTypeStep].ApproximateBacklogCount)

Connecting

gush.Options is the SDK's client.Options, including TLS, namespaces and credentials. For Temporal Cloud with an API key:

options := gush.Options{
    HostPort:    "my-namespace.a1b2c.tmprl.cloud:7233",
    Namespace:   "my-namespace.a1b2c",
    Credentials: client.NewAPIKeyStaticCredentials(apiKey),
}
c, err := gush.Dial(options)

An empty HostPort falls back to TEMPORAL_ADDRESS, then localhost:7233. An empty Namespace falls back to TEMPORAL_NAMESPACE, then the default namespace.

The package-level functions, such as gush.RunFlow and gush.Serve, use a shared client configured from those variables. They suit a process that talks to one server.

Contributing

See CONTRIBUTING.md and SECURITY.md.

License

MIT

Documentation

Overview

Package gush is a small Go layer over Temporal: write flows and steps as plain functions, and run them on a Temporal server.

A flow is deterministic orchestration, a Temporal workflow. A step is the unit of real work, a Temporal activity: network calls, database writes and computation belong there, because Temporal replays flows and retries steps. 

func Charge(ctx *gush.StepContext) (any, error) {
    var order Order
    err := ctx.Decode(&order)
    if err != nil {
        return nil, err
    }
    return charge(ctx, order)
}

func Fulfill(ctx *gush.FlowContext) (any, error) {
    var receipt Receipt
    err := ctx.RunStep(Charge, ctx.Input, &receipt, gush.WithMaxAttempts(3))
    return receipt, err
}

A StepContext is a standard context.Context. A FlowContext holds the Temporal SDK's workflow.Context in its Context field, for SDK functions such as workflow.Sleep.

Index

Constants

View Source 
const DefaultFlowVersion = int(workflow.DefaultVersion)

DefaultFlowVersion is what Version returns for an execution that passed the call before it was added.

View Source 
const StepTimeoutErrorType = "TemporalTimeout:StartToClose"

StepTimeoutErrorType is the error type of an attempt that exceeded its timeout. Pass it to WithNoRetry to stop retrying timed-out steps.

View Source 
const UnlimitedAttempts int32 = -1

UnlimitedAttempts as MaxAttempts keeps retrying until something else ends the run, such as a timeout.

Variables

This section is empty.

Functions

func CancelFlow 

func CancelFlow(ctx context.Context, flowID string) error

CancelFlow calls CancelFlow on the default client.

func Close 

func Close()

Close closes the default client if it was connected. A later package-level call connects again.

func Decode 

func Decode(data []byte, target any) error

Decode deserializes data produced by Encode into target. A pointer to anything Encode uses as it is, through any number of pointers, receives the bytes as they are, empty data zeroes target, and everything else is decoded as JSON. *any receives what the JSON holds, or a string when the data is not JSON.

func DecodePayload 

func DecodePayload(payload []byte, target any) error

DecodePayload reverses EncodePayload. An empty payload or nil target leaves target alone.

func Encode 

func Encode(value any) ([]byte, error)

Encode serializes a value the way flows and steps do for their input and output. Strings and byte slices are used as they are, through any number of pointers and including types built on them, unless the type has its own JSON or text methods. A nil value or nil pointer encodes to nil, and everything else becomes JSON without HTML escaping.

Adding or removing those methods changes how a type built on a string or bytes encodes, so a flow in flight then reads its old value wrongly. Drain such flows first.

func EncodePayload 

func EncodePayload(value any) ([]byte, error)

EncodePayload is the wire form of a flow or step input or output: Encode, then zstd. Code that works below Gush needs it, such as a test that starts a flow through the SDK's testsuite.

func GetQueueStats 

func GetQueueStats(ctx context.Context, queue string) (map[QueueType]*QueueStats, error)

GetQueueStats calls GetQueueStats on the default client.

func InitLogExport 

func InitLogExport() (shutdown func(), err error)

InitLogExport starts exporting logs over OTLP when OTEL_EXPORTER_OTLP_LOGS_ENDPOINT or OTEL_EXPORTER_OTLP_ENDPOINT is set, and returns a function that flushes them and stops. The endpoint must be an http:// or https:// URL. OpenTelemetry's other standard variables set the protocol, headers and TLS, and the default protocol is http/protobuf. Dial calls it, so call it directly only for a process that logs without a Client.

A LoggerProvider the host already set with OpenTelemetry's global.SetLoggerProvider always wins, and InitLogExport leaves it alone. Importing gush never changes the global provider; InitLogExport sets it only when it starts exporting. With no provider and no endpoint, logs go to standard error.

Export is shared by the whole process: it stops when the last returned function has run.

func IsApplicationError 

func IsApplicationError(err error) bool

IsApplicationError reports whether err is an ApplicationError.

func IsFlowRunningError 

func IsFlowRunningError(err error) bool

IsFlowRunningError reports whether a flow did not start because one with the same ID is running. A child flow gets this error, and so does a client using WithErrorIfRunning. A step using such a client must check the error itself, because a flow cannot recognize it after it crosses the step.

func IsInvalidConfigError added in v0.2.0 

func IsInvalidConfigError(err error) bool

IsInvalidConfigError reports whether a request was refused as invalid, so sending it again cannot succeed: Gush found settings Temporal refuses, including in a flow that has since failed on them, or the server answered InvalidArgument. The server gives that answer for more than settings, such as an input over its size limit. Past a flow it sees through Abort, NewPermanentError and %w, but not errors.Join or several %w, and not a client's refusal inside a step, whose text is all that crosses the step.

func IsPanicError 

func IsPanicError(err error) bool

IsPanicError reports whether err is a PanicError.

func IsPermanentError 

func IsPermanentError(err error) bool

IsPermanentError reports whether err is an ApplicationError that is not retried.

func IsResumeTokenError 

func IsResumeTokenError(err error) bool

IsResumeTokenError reports whether Resume failed because its token is invalid, or stale as IsStaleResumeTokenError describes.

func IsStaleResumeTokenError added in v0.2.0 

func IsStaleResumeTokenError(err error) bool

IsStaleResumeTokenError reports whether Resume failed because its token's attempt is over: the step ended, or it is retrying under a new token. That makes it a late callback rather than a bad one, though the server cannot tell it from a well-formed token that matches no step, such as a corrupted one. Every such error is also an IsResumeTokenError.

func NewLogExporter added in v0.2.0 

func NewLogExporter(ctx context.Context) (sdklog.Exporter, error)

NewLogExporter creates the OTLP exporter InitLogExport uses, from the same variables, for a process that builds its own LoggerProvider. It returns nil when no endpoint is configured. The exporter reads OpenTelemetry's standard variables itself. It accepts a malformed endpoint, such as collector:4318, and then loses every record, so that is an error here.

func Resume 

func Resume(ctx context.Context, resumeToken string, result any, stepErr error) error

Resume calls Resume on the default client.

func RunFlow 

func RunFlow(ctx context.Context, request FlowRequest, options ...FlowOption) error

RunFlow calls RunFlow on the default client.

func Serve 

func Serve(workers ...*Worker) error

Serve calls Serve on the default client.

func StartFlow 

func StartFlow(ctx context.Context, request FlowRequest, options ...FlowOption) (string, error)

StartFlow calls StartFlow on the default client.

func TerminateFlow 

func TerminateFlow(ctx context.Context, flowID, reason string) error

TerminateFlow calls TerminateFlow on the default client.

Types

type ApplicationError 

type ApplicationError = temporal.ApplicationError

ApplicationError is an error with a type name that retry policies can match.

func NewApplicationError 

func NewApplicationError(message, errType string, details ...any) *ApplicationError

NewApplicationError creates a retryable error. Pass errType to WithNoRetry to stop retrying errors of this type on a particular run.

func NewPermanentError 

func NewPermanentError(message, errType string, cause error, details ...any) *ApplicationError

NewPermanentError creates an error that fails its flow or step immediately, whatever the retry policy says.

type Client 

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

Client is a connection to a Temporal server.

func DefaultClient 

func DefaultClient() (*Client, error)

DefaultClient returns the process-wide client behind the package-level functions, connecting on first use with the settings Dial reads from the environment. A failed connection is not remembered, so the next call tries again.

func Dial 

func Dial(options Options) (*Client, error)

Dial connects to Temporal and starts OTLP log export when it is configured; see InitLogExport.

An empty HostPort falls back to TEMPORAL_ADDRESS, then the SDK's default of localhost:7233. An empty Namespace falls back to TEMPORAL_NAMESPACE, then the SDK's default namespace.

func (*Client) CancelFlow 

func (c *Client) CancelFlow(ctx context.Context, flowID string) error

CancelFlow asks a flow to stop. Its steps and the children it waits for are canceled too, so a RunStep in progress returns a canceled error, or waits first with WithWaitForCancel. A child begun with StartFlow is independent and keeps running. Cleanup steps need a disconnected context; see FlowContext.WithContext. A step that is already running only finds out when it calls Heartbeat, and WithHeartbeatTimeout makes that prompt.

func (*Client) Close 

func (c *Client) Close()

Close closes the connection and flushes the logs this client started exporting. It is safe to call more than once.

func (*Client) GetFlow 

func (c *Client) GetFlow(ctx context.Context, flowID string) (*FlowInfo, error)

GetFlow describes a flow execution.

func (*Client) GetQueueStats 

func (c *Client) GetQueueStats(ctx context.Context, queue string) (map[QueueType]*QueueStats, error)

GetQueueStats returns a queue's load for each task type. 

stats, err := gush.GetQueueStats(ctx, "my-queue")
if err != nil {
    return err
}
log.Printf("step backlog: %d", stats[gush.QueueTypeStep].ApproximateBacklogCount)

func (*Client) Resume 

func (c *Client) Resume(ctx context.Context, resumeToken string, result any, stepErr error) error

Resume finishes a suspended step with a result or, if stepErr is set, an error. See StepContext.Suspend.

An error fails the step for good, because running its code again would repeat whatever it asked for, such as a person's approval. To run the code again under the step's retry policy, pass a NewApplicationError, which is a request to retry. A canceled error cancels the step if it is being canceled, and otherwise fails it for good too. An ApplicationError caused by a cancellation is recorded without that cause or its details.

func (*Client) RunFlow 

func (c *Client) RunFlow(ctx context.Context, request FlowRequest, options ...FlowOption) error

RunFlow starts a flow and waits for its result.

func (*Client) Serve 

func (c *Client) Serve(workers ...*Worker) error

Serve runs workers until the process receives SIGINT or SIGTERM or every worker's context is canceled, then returns once they have stopped. It returns an error if the workers are misconfigured or one fails, and a failed worker stops the others.

func (*Client) StartFlow 

func (c *Client) StartFlow(ctx context.Context, request FlowRequest, options ...FlowOption) (string, error)

StartFlow starts a flow and returns its ID without waiting for it to finish.

func (*Client) TerminateFlow 

func (c *Client) TerminateFlow(ctx context.Context, flowID, reason string) error

TerminateFlow stops a flow immediately, with no cleanup. Prefer CancelFlow.

type CommonConfig 

type CommonConfig struct {
	RetryPolicy

	// Priority controls scheduling order; higher priority runs first.
	Priority Priority `json:"priority"`

	// FairKey groups work for fair scheduling across tenants or user groups. Keys are
	// cut to the 64 bytes Temporal accepts, so keys that agree up to there share a group.
	FairKey string `json:"fair_key"`

	// LogAttributes are attached to every log record and inherited by child flows and steps.
	LogAttributes LogAttributeMap `json:"log_attributes"`
}

CommonConfig holds the settings flows and steps share.

type FlowConfig 

type FlowConfig struct {
	CommonConfig

	// ID is the flow ID. If empty, one is generated.
	ID string `json:"id"`

	// Timeout limits the whole execution, including every retry. Zero means unlimited, and a
	// negative one is an error.
	Timeout time.Duration `json:"timeout"`

	// ErrorIfRunning makes a client's RunFlow or StartFlow fail rather than join a flow
	// already running with this ID. See WithErrorIfRunning.
	ErrorIfRunning bool `json:"error_if_running"`
}

FlowConfig holds the settings of one flow run. Its JSON form is stable, so a config can be queued and applied later with WithConfig. A setting left out of the JSON, or zero, uses its default, "max_attempts": -1 means unlimited, and durations are integer nanoseconds. Waiting is not part of a config: whoever applies it chooses RunFlow or StartFlow.

func DefaultFlowConfig 

func DefaultFlowConfig() *FlowConfig

DefaultFlowConfig returns the settings a flow run starts from: one attempt, medium priority, and a 1s to 100s backoff if more attempts are allowed.

func (*FlowConfig) Apply 

func (config *FlowConfig) Apply(options ...FlowOption)

Apply applies options to config in order.

type FlowContext 

type FlowContext struct {
	// Context is the Temporal workflow context that RunStep and RunFlow run on.
	Context workflow.Context

	FlowInfo
	*Logger

	// Input is the raw input; Decode deserializes it.
	Input []byte

	// LogAttributes are inherited from the parent and passed on to child flows and steps.
	LogAttributes map[string]any
}

FlowContext is what a flow receives. Its Context field is the Temporal SDK's own context, for SDK functions such as workflow.Sleep(ctx.Context, d).

Temporal replays a flow's code to restore it, so flow code must be deterministic: no I/O, no time.Now and no time.Sleep. Do real work in steps. Logs written during a replay are dropped, so each line appears once. 

func OrderFulfillment(ctx *gush.FlowContext) (any, error) {
    var order Order
    err := ctx.Decode(&order)
    if err != nil {
        return nil, err
    }

    var payment Payment
    err = ctx.RunStep(ProcessPayment, order, &payment)
    if err != nil {
        return nil, err
    }

    var receipt Receipt
    err = ctx.RunStep(ShipOrder, payment, &receipt)
    return receipt, err
}

func NewFlowContext 

func NewFlowContext(ctx workflow.Context, logAttributes map[string]any) *FlowContext

NewFlowContext creates a FlowContext from a Temporal workflow context.

func (*FlowContext) Abort 

func (flowCtx *FlowContext) Abort(cause ...error) error

Abort returns an error that fails the flow permanently, with no further retries.

func (*FlowContext) ContinueAsNew 

func (flowCtx *FlowContext) ContinueAsNew(input any) error

ContinueAsNew returns an error that ends this run and starts the same flow again with the given input, the same flow ID and log attributes, and an empty history. Temporal limits a flow's history, so a flow that loops for a long time returns this every so often. 

if batch.Done {
    return summary, nil
}
return nil, ctx.ContinueAsNew(batch.Next)

func (*FlowContext) Decode 

func (flowCtx *FlowContext) Decode(target any) error

Decode deserializes the flow's Input into target, aborting the flow if it cannot.

func (*FlowContext) RunFlow 

func (flowCtx *FlowContext) RunFlow(request FlowRequest, options ...FlowOption) error

RunFlow runs a child flow and waits for its result. The child inherits this flow's priority, fair key and log attributes unless options override them.

func (*FlowContext) RunStep 

func (flowCtx *FlowContext) RunStep(step any, input, output any, options ...StepOption) error

RunStep runs a step and waits for its result. The step inherits this flow's priority, fair key and log attributes unless options override them.

func (*FlowContext) StartFlow 

func (flowCtx *FlowContext) StartFlow(request FlowRequest, options ...FlowOption) (string, error)

StartFlow starts a child flow and returns its ID once it has started, without waiting for it to finish. The child is independent: it keeps running when this flow ends or is canceled, and CancelFlow stops it by that ID. It inherits this flow's priority, fair key and log attributes unless options override them. It returns a canceled error if its own context is already canceled.

func (*FlowContext) Version 

func (flowCtx *FlowContext) Version(tag string, minVersion, maxVersion int) int

Version lets flow code change without breaking executions already running the old code. The first execution of a call records maxVersion in the flow's history, and replays return that recorded value. An execution that passed this point before the call existed gets DefaultFlowVersion, and one that reaches it later records maxVersion however long ago it started.

To replace a step: 

v := ctx.Version("payment-provider", gush.DefaultFlowVersion, 1)
if v == gush.DefaultFlowVersion {
    err = ctx.RunStep(OldCharge, input, &output)
} else {
    err = ctx.RunStep(NewCharge, input, &output)
}

To change the same code again, keep the tag, raise maxVersion and add a branch. Once no execution can reach an old branch, delete it, keep the Version call and raise minVersion to the oldest version left. An execution that recorded an older version then stops making progress, with its flow tasks failing, until code that supports it is deployed. It never runs a branch it did not take.

func (*FlowContext) WithContext 

func (flowCtx *FlowContext) WithContext(ctx workflow.Context) *FlowContext

WithContext returns a copy of this FlowContext that runs its steps and child flows on ctx. RunStep and RunFlow always use the FlowContext's own workflow.Context, so code holding a different one needs this, such as a goroutine started with workflow.Go or cleanup on a workflow.NewDisconnectedContext. StartFlow only checks that context for cancellation, because its child is independent. 

workflow.Go(ctx.Context, func(g workflow.Context) {
    err := ctx.WithContext(g).RunStep(Process, item, nil)
    ...
})

type FlowHandler 

type FlowHandler struct {
	Name string
	Func FlowHandlerFunc
}

FlowHandler is a flow registered under an explicit name.

func NewFlowHandler 

func NewFlowHandler(name string, fn FlowHandlerFunc) *FlowHandler

NewFlowHandler names a flow. Use it when the function's own name is not unique or not stable, such as for a closure. 

handler := gush.NewFlowHandler("data-pipeline", DataPipeline)
worker := gush.NewWorker("my-queue").WithFlowHandlers(handler)

type FlowHandlerFunc 

type FlowHandlerFunc func(ctx *FlowContext) (output any, err error)

FlowHandlerFunc is the signature of a flow.

type FlowInfo 

type FlowInfo struct {
	ID       string // generated if not set at start
	Status   string // Running, Completed, Failed, Canceled, Terminated, TimedOut or ContinuedAsNew
	ParentID string // empty for a top-level flow
	Queue    string
	Priority Priority
	FairKey  string

	// Attempt counts from 1. It is set inside a running flow; GetFlow leaves it zero
	// because Temporal does not report it there.
	Attempt int32

	StartTime *time.Time
	CloseTime *time.Time // nil while running
}

FlowInfo describes a flow execution.

func GetFlow 

func GetFlow(ctx context.Context, flowID string) (*FlowInfo, error)

GetFlow calls GetFlow on the default client.

type FlowOption 

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

FlowOption customizes a flow run.

func WithConfig 

func WithConfig(config *FlowConfig) FlowOption

WithConfig replaces the settings accumulated so far with a copy of config, so options after it refine config. An empty Priority or FairKey keeps the current one, and LogAttributes merge over the current ones, so a child flow keeps what it inherits. ErrorIfRunning stays on once set, because a config cannot switch off that safety check. A nil config changes nothing.

func WithErrorIfRunning 

func WithErrorIfRunning() FlowOption

WithErrorIfRunning makes a client's RunFlow or StartFlow return an error, which IsFlowRunningError recognizes, rather than join a flow already running with this ID.

func WithID 

func WithID(flowID string) FlowOption

WithID sets the flow ID. Without it one is generated, which StartFlow returns. While a flow with this ID is running, a client's RunFlow or StartFlow joins it instead of starting another and ignores this run's flow, input and settings, so concurrent callers can share one run. RunFlow then returns that flow's result. Use WithErrorIfRunning to be told instead. A child flow always gets an error, because a flow cannot wait for a flow it did not start; check it with IsFlowRunningError.

func WithTimeout 

func WithTimeout(timeout time.Duration) FlowOption

WithTimeout limits the whole flow execution, including every retry. A negative timeout is an error.

type FlowRequest 

type FlowRequest struct {
	// Queue is the task queue to run on. A client must set it; a child flow left empty
	// runs on its parent's queue.
	Queue string

	// Flow is the flow function or the name it is registered under.
	Flow any

	// Input is the flow's input. See Encode for how it is serialized.
	Input any

	// Output receives the result of RunFlow, unless it is nil. StartFlow does not use it.
	Output any
}

FlowRequest says which flow to run, where, and with what data.

type LogAttribute 

type LogAttribute = attribute.KeyValue

LogAttribute is a key and value on a log record. It is OpenTelemetry's attribute.KeyValue, so attribute.Int64Slice and the like work wherever one is accepted.

func Attr 

func Attr(key string, value any) LogAttribute

Attr creates a log attribute. Strings, booleans, numbers and []byte keep their type, a nil value has no value, and anything else is recorded in its fmt.Sprint form.

type LogAttributeMap added in v0.2.0 

type LogAttributeMap map[string]any

LogAttributeMap holds the fields attached to every log record of a run, by key.

func (*LogAttributeMap) UnmarshalJSON added in v0.2.0 

func (attributes *LogAttributeMap) UnmarshalJSON(data []byte) error

UnmarshalJSON adds the attributes in data to these, keeping numbers exact, and null clears them. Plain JSON decoding would turn every number into a float64, which damages large IDs.

type Logger 

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

Logger writes structured log records with timing helpers. Flows and steps embed one, so ctx.Printf works directly in a handler, and records carry flow_id and the inherited log attributes wherever the flow's work runs.

See InitLogExport for where records go.

func NewLogger 

func NewLogger(attributes ...LogAttribute) *Logger

NewLogger creates a Logger whose records carry the given attributes.

func (*Logger) Print 

func (logger *Logger) Print(v ...any)

Print logs its arguments in the manner of fmt.Print.

func (*Logger) PrintWithAttributes 

func (logger *Logger) PrintWithAttributes(message string, attributes ...LogAttribute)

PrintWithAttributes logs a message with attributes on this record only.

func (*Logger) Printf 

func (logger *Logger) Printf(format string, v ...any)

Printf logs its arguments in the manner of fmt.Printf.

func (*Logger) Println 

func (logger *Logger) Println(v ...any)

Println logs its arguments in the manner of fmt.Println, without the trailing newline.

func (*Logger) SetAttributes 

func (logger *Logger) SetAttributes(attributes ...LogAttribute)

SetAttributes adds attributes to every later record, replacing any with the same key.

func (*Logger) Time 

func (logger *Logger) Time(label string)

Time starts a named timer for TimeEnd.

func (*Logger) TimeEnd 

func (logger *Logger) TimeEnd(label string, v ...any)

TimeEnd logs the time since the matching Time call, followed by any extra values, and removes the timer.

func (*Logger) Write 

func (logger *Logger) Write(p []byte) (n int, err error)

Write logs each non-empty line of p, so a Logger can receive the standard log package's output.

type Option 

type Option interface {
	FlowOption
	StepOption
}

Option customizes a flow run or a step run.

func WithBackoffCoefficient 

func WithBackoffCoefficient(coefficient float64) Option

WithBackoffCoefficient sets the multiplier applied to the delay after each retry. It must be at least 1.

func WithFairKey 

func WithFairKey(key string) Option

WithFairKey sets the group this work is scheduled fairly against, such as a tenant ID. Keys are cut to the 64 bytes Temporal accepts, so keys that agree up to there share a group.

func WithInitialInterval 

func WithInitialInterval(interval time.Duration) Option

WithInitialInterval sets the delay before the first retry.

func WithLogAttribute 

func WithLogAttribute(key string, value any) Option

WithLogAttribute adds one attribute to every log record of this run and its children.

func WithLogAttributes 

func WithLogAttributes(attributes map[string]any) Option

WithLogAttributes merges attributes into every log record of this run and its children.

func WithMaxAttempts 

func WithMaxAttempts(attempts int32) Option

WithMaxAttempts sets the total number of attempts, including the first. Zero means one for a flow and ten for a step. Use UnlimitedAttempts to keep retrying. Any other negative number makes RunFlow, StartFlow or RunStep return an error.

func WithMaxInterval 

func WithMaxInterval(interval time.Duration) Option

WithMaxInterval caps the delay between retries.

func WithNoRetry 

func WithNoRetry(errorTypes ...string) Option

WithNoRetry adds ApplicationError types that fail immediately instead of retrying. See RetryPolicy.NoRetry for entries that name a timeout, where an unknown name is an error.

func WithNoRetryErrors 

func WithNoRetryErrors(errs ...*ApplicationError) Option

WithNoRetryErrors adds the types of the given errors to the ones that fail immediately.

func WithPriority 

func WithPriority(priority Priority) Option

WithPriority sets the scheduling priority.

func WithRetryPolicy 

func WithRetryPolicy(policy RetryPolicy) Option

WithRetryPolicy replaces the whole retry policy.

type Options 

type Options = client.Options

Options configures the connection to Temporal, including its address, namespace, TLS and credentials. It is the Temporal SDK's client.Options.

type PanicError 

type PanicError = temporal.PanicError

PanicError is an unrecovered panic during a step.

type Priority 

type Priority string

Priority is the scheduling priority of a flow or step. Higher priority runs first. 

const (
	PriorityUrgent  Priority = "urgent"
	PriorityHigh    Priority = "high"
	PriorityMedium  Priority = "medium" // default
	PriorityLow     Priority = "low"
	PriorityTrivial Priority = "trivial"
)

func NewPriority 

func NewPriority(key int) Priority

NewPriority converts a Temporal priority key back to a Priority. An unknown key counts as PriorityMedium.

func (Priority) Int 

func (priority Priority) Int() int

Int returns the Temporal priority key, from 1 for PriorityUrgent to 5 for PriorityTrivial. An unknown priority counts as PriorityMedium.

type QueueStats 

type QueueStats struct {
	// ApproximateBacklogCount is about how many tasks are waiting. It may count expired
	// tasks but converges to the right value.
	ApproximateBacklogCount int64

	// ApproximateBacklogAge is about how long the oldest waiting task has been queued.
	ApproximateBacklogAge time.Duration

	// BacklogIncreaseRate is TasksAddRate minus TasksDispatchRate. Positive means the
	// backlog is growing.
	BacklogIncreaseRate float32

	// TasksAddRate is about how many tasks per second are added, over the last 30 seconds.
	TasksAddRate float32

	// TasksDispatchRate is about how many tasks per second reach workers, over the last
	// 30 seconds.
	TasksDispatchRate float32
}

QueueStats reports the load on a task queue, for monitoring and scaling decisions.

type QueueType 

type QueueType = client.TaskQueueType

QueueType separates a queue's flow tasks from its step tasks. 

const (
	QueueTypeFlow QueueType = client.TaskQueueTypeWorkflow
	QueueTypeStep QueueType = client.TaskQueueTypeActivity
)

type RetryPolicy 

type RetryPolicy struct {
	// InitialInterval is the delay before the first retry. Zero means 1s.
	InitialInterval time.Duration `json:"initial_interval"`

	// BackoffCoefficient multiplies the delay after each retry. It must be at least 1.
	// Zero means 2.0.
	BackoffCoefficient float64 `json:"backoff_coefficient"`

	// MaxInterval caps the delay between retries. Zero means 100 times InitialInterval, and a
	// cap below InitialInterval is raised to it.
	MaxInterval time.Duration `json:"max_interval"`

	// MaxAttempts is the total number of attempts, including the first. Zero means one for
	// a flow and ten for a step. Use UnlimitedAttempts to keep retrying. Any other negative
	// number is an error.
	MaxAttempts int32 `json:"max_attempts"`

	// NoRetry lists the ApplicationError types that fail immediately. An entry starting with
	// TemporalTimeout: names a timeout instead, such as StepTimeoutErrorType. Naming a timeout
	// Temporal does not know is an error, unless there is only one attempt and so no retry.
	// Use the short name: Temporal also accepts TIMEOUT_TYPE_START_TO_CLOSE, which never matches.
	NoRetry []string `json:"no_retry"`
}

RetryPolicy defines how flows and steps are retried after they fail. Every setting left at zero uses its default.

type StderrLogExporter 

type StderrLogExporter struct{}

StderrLogExporter writes log records to standard error, one line per record. Gush uses it when no other destination is configured, and it can back a host's own LoggerProvider.

func NewStderrLogExporter 

func NewStderrLogExporter() *StderrLogExporter

NewStderrLogExporter creates a StderrLogExporter.

func (*StderrLogExporter) Export 

func (exporter *StderrLogExporter) Export(ctx context.Context, records []sdklog.Record) error

Export writes records until ctx is canceled.

func (*StderrLogExporter) ForceFlush 

func (exporter *StderrLogExporter) ForceFlush(context.Context) error

ForceFlush does nothing.

func (*StderrLogExporter) Shutdown 

func (exporter *StderrLogExporter) Shutdown(context.Context) error

Shutdown does nothing.

type StepConfig 

type StepConfig struct {
	CommonConfig

	// Queue is the task queue the step runs on. If empty, it uses its flow's queue.
	Queue string `json:"queue"`

	// AttemptTimeout limits each attempt, not the step as a whole. It also bounds how long
	// a suspended step waits for Resume. A negative one is an error, and so is zero, unless
	// PickupTimeout and a limited number of attempts give the step an overall limit.
	AttemptTimeout time.Duration `json:"attempt_timeout"`

	// HeartbeatTimeout fails an attempt that goes this long without calling Heartbeat.
	// Zero disables it, and a negative one is an error.
	HeartbeatTimeout time.Duration `json:"heartbeat_timeout"`

	// PickupTimeout fails the step if no worker picks it up in time. Zero or negative means
	// unlimited.
	PickupTimeout time.Duration `json:"pickup_timeout"`

	// WaitForCancel makes RunStep wait for a canceled step to end. See WithWaitForCancel.
	WaitForCancel bool `json:"wait_for_cancel"`
}

StepConfig holds the settings of one step run.

func DefaultStepConfig 

func DefaultStepConfig() *StepConfig

DefaultStepConfig returns the settings a step run starts from: ten attempts of up to 2 minutes each, medium priority, and a 1s to 100s backoff.

func (*StepConfig) Apply 

func (config *StepConfig) Apply(options ...StepOption)

Apply applies options to config in order.

type StepContext 

type StepContext struct {
	context.Context
	StepInfo
	*Logger

	// Input is the raw input; Decode deserializes it.
	Input []byte

	// LogAttributes are inherited from the parent flow.
	LogAttributes map[string]any
}

StepContext is what a step receives. It is a context.Context, so pass it to anything that takes one, and the step's cancellation and deadline follow. 

func ProcessOrder(ctx *gush.StepContext) (any, error) {
    var order Order
    err := ctx.Decode(&order)
    if err != nil {
        return nil, err
    }
    return processOrder(ctx, order)
}

func AsStepContext 

func AsStepContext(ctx context.Context) *StepContext

AsStepContext returns ctx as a StepContext, so helpers that take a plain context.Context can log with the step's context. A context derived from a StepContext, such as with context.WithTimeout, keeps ctx's cancellation and the step's logger and attributes. Any other context is wrapped.

func NewStepContext 

func NewStepContext(ctx context.Context, logAttributes map[string]any) *StepContext

NewStepContext creates a StepContext from any context. Outside a running step, such as in a test, its step information is empty.

func (*StepContext) Abort 

func (stepCtx *StepContext) Abort(cause ...error) error

Abort returns an error that fails the step permanently, with no further retries.

func (*StepContext) Decode 

func (stepCtx *StepContext) Decode(target any) error

Decode deserializes the step's Input into target, aborting the step if it cannot.

func (*StepContext) Heartbeat 

func (stepCtx *StepContext) Heartbeat(details ...any)

Heartbeat reports that the step is alive, which WithHeartbeatTimeout requires. It is also how a running step learns of cancellation: ctx is canceled only after a heartbeat. Details are recorded for the next attempt to read with the SDK's activity.GetHeartbeatDetails. Outside a running step it does nothing.

func (*StepContext) Suspend 

func (stepCtx *StepContext) Suspend() (any, error)

Suspend ends the attempt without finishing the step, freeing the worker while something outside the flow, such as a person or a webhook, decides the result and calls Resume with ctx.ResumeToken.

The attempt timeout keeps running while the step waits. If it expires first, the step retries and runs again, which by default happens after 2 minutes. So set WithAttemptTimeout to the longest acceptable wait, and WithMaxAttempts(1) unless running the step's code twice is safe. 

func ApproveExpense(ctx *gush.StepContext) (any, error) {
    err := sendApprovalRequest(ctx, ctx.ResumeToken, ctx.Input)
    if err != nil {
        return nil, err
    }
    return ctx.Suspend()
}

err := ctx.RunStep(ApproveExpense, expense, &decision,
    gush.WithAttemptTimeout(24*time.Hour),
    gush.WithMaxAttempts(1),
)

type StepHandler 

type StepHandler struct {
	Name string
	Func StepHandlerFunc
}

StepHandler is a step registered under an explicit name.

func NewStepHandler 

func NewStepHandler(name string, fn StepHandlerFunc) *StepHandler

NewStepHandler names a step. Use it when the function's own name is not unique or not stable, such as for a closure. 

handler := gush.NewStepHandler("process-data", ProcessData)
worker := gush.NewWorker("my-queue").WithStepHandlers(handler)

type StepHandlerFunc 

type StepHandlerFunc func(ctx *StepContext) (output any, err error)

StepHandlerFunc is the signature of a step. Temporal limits its encoded output to 2MB, so return a reference to anything larger.

type StepInfo 

type StepInfo struct {
	// ID identifies this step within its flow.
	ID string

	// FlowID is the flow that started this step.
	FlowID string

	// Queue is the task queue this step runs on.
	Queue string

	// Priority is this step's scheduling priority.
	Priority Priority

	// FairKey is this step's fairness group.
	FairKey string

	// ResumeToken is what Resume needs to finish this step after it suspends. It belongs to
	// this attempt, and a retry gets a new one.
	ResumeToken string

	// Attempt counts from 1.
	Attempt int32
}

StepInfo describes a step execution.

type StepOption 

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

StepOption customizes a step run.

func WithAttemptTimeout 

func WithAttemptTimeout(timeout time.Duration) StepOption

WithAttemptTimeout limits each attempt of the step. See StepConfig.AttemptTimeout for what is an error.

func WithHeartbeatTimeout 

func WithHeartbeatTimeout(timeout time.Duration) StepOption

WithHeartbeatTimeout fails an attempt that goes this long without calling Heartbeat. A negative timeout is an error.

func WithPickupTimeout 

func WithPickupTimeout(timeout time.Duration) StepOption

WithPickupTimeout fails the step if no worker picks it up in time.

func WithQueue 

func WithQueue(queue string) StepOption

WithQueue runs the step on another task queue.

func WithWaitForCancel 

func WithWaitForCancel() StepOption

WithWaitForCancel makes RunStep wait for a running step to end when the flow is canceled. Otherwise it returns a canceled error at once while the step carries on, so cleanup could overlap the step's work. The step ends sooner if it heartbeats and returns on ctx.Done(), and if it finishes anyway, RunStep returns its result.

type Worker 

type Worker struct {
	Queue string

	// StopTimeout is how long running tasks get to finish on shutdown. Default: 30s.
	StopTimeout time.Duration

	// QueueRate is the server-side limit of steps per second, across all workers on this
	// queue. Default: 100,000.
	QueueRate float64

	// StepRate is the limit of steps per second in this worker. Default: 100,000.
	StepRate float64

	// MaxSteps is how many steps this worker runs at once. Default: 1,000.
	MaxSteps int

	// StepPollers is how many goroutines fetch steps. Default: 2.
	StepPollers int

	// MaxFlowTasks is how many flow tasks this worker runs at once. A flow only occupies a
	// task while it advances, so far more flows than this can be in progress. Default: 1,000.
	MaxFlowTasks int

	// FlowPollers is how many goroutines fetch flow tasks. Default: 2.
	FlowPollers int

	// Steps and Flows are the registered handlers.
	Steps []*StepHandler
	Flows []*FlowHandler

	// Context stops the worker when canceled.
	Context context.Context
}

Worker configures a process's work on one task queue. NewWorker sets the defaults.

func NewWorker 

func NewWorker(queue string) *Worker

NewWorker creates a Worker for a queue.

func (*Worker) WithContext 

func (w *Worker) WithContext(ctx context.Context) *Worker

WithContext sets the context that stops the worker when canceled.

func (*Worker) WithFlowHandlers 

func (w *Worker) WithFlowHandlers(flows ...*FlowHandler) *Worker

WithFlowHandlers registers flows under explicit names.

func (*Worker) WithFlowPollers 

func (w *Worker) WithFlowPollers(count int) *Worker

WithFlowPollers sets how many goroutines fetch flow tasks. Temporal rejects 1, which Serve reports. Zero uses the SDK's default.

func (*Worker) WithFlows 

func (w *Worker) WithFlows(flows ...FlowHandlerFunc) *Worker

WithFlows registers flows under their Go function names, without package or receiver.

func (*Worker) WithMaxFlowTasks 

func (w *Worker) WithMaxFlowTasks(limit int) *Worker

WithMaxFlowTasks sets how many flow tasks this worker runs at once. Temporal needs at least 2, so a lower value counts as 2.

func (*Worker) WithMaxSteps 

func (w *Worker) WithMaxSteps(limit int) *Worker

WithMaxSteps sets how many steps this worker runs at once.

func (*Worker) WithQueueRate 

func (w *Worker) WithQueueRate(rate float64) *Worker

WithQueueRate sets the server-side limit of steps per second across this queue's workers.

func (*Worker) WithStepHandlers 

func (w *Worker) WithStepHandlers(steps ...*StepHandler) *Worker

WithStepHandlers registers steps under explicit names.

func (*Worker) WithStepPollers 

func (w *Worker) WithStepPollers(count int) *Worker

WithStepPollers sets how many goroutines fetch steps.

func (*Worker) WithStepRate 

func (w *Worker) WithStepRate(rate float64) *Worker

WithStepRate sets the limit of steps per second in this worker.

func (*Worker) WithSteps 

func (w *Worker) WithSteps(steps ...StepHandlerFunc) *Worker

WithSteps registers steps under their Go function names, without package or receiver.

func (*Worker) WithStopTimeout 

func (w *Worker) WithStopTimeout(timeout time.Duration) *Worker

WithStopTimeout sets how long running tasks get to finish on shutdown.

Source Files

View all Source files

Directories

Path Synopsis
examples
quickstart command
Quickstart runs a worker and one flow in the same process.
 Quickstart runs a worker and one flow in the same process.

Jump to

Keyboard shortcuts

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