README
¶
Memory Workflow
arkoren.dev/core/service/workflow/memoryworkflow is an in-process implementation of the Arkoren
workflow.Driver contract for local development and single-replica
deployments. It runs every workflow in a goroutine of the process and uses the
system clock. Runs are lost on restart and are not shared between replicas.
It is included in go get arkoren.dev/core. Import
"arkoren.dev/core/service/workflow" and
"arkoren.dev/core/service/workflow/memoryworkflow". The
memory workflow guide
covers cancellation, signal limits, and cleanup ownership.
Declare workflows with workflow.New(name, fn, configure...) and steps with
workflow.NewStep(name, fn, configure...). Configure
workflow.Builder.MaxDuration before registration; omission fails in
workflow.Handlers. Definition callbacks publish detached settings, and
Step.Derive supplies a local policy override without changing the base step.
Given the onboarding definitions, construct the process-local driver inside an application composer and register cleanup for its runs:
registry := compose.Must(workflow.Handlers(Onboard, SendWelcome))
driver := memoryworkflow.New(registry)
compose.Cleanup(driver.Stop)
client := workflow.NewClient(driver)
Here compose is the *application.Composer passed to your bootstrap function.
Pass client explicitly to the application services that need it.
Context.Child checks for an existing run and inserts a new run under one lock.
Only a newly created child's context derives from the parent; it stops when the
parent is canceled or times out. Joining an existing run ignores replacement
input and acquires no cancellation ownership. Ending that joining parent's wait
leaves the original run active and able to finish through a signal.
To test code that starts workflows, use workflowtest.Fake. To test the
workflows themselves with sleeps and timeouts, run the driver inside
testing/synctest, which makes time pass instantly.
Documentation
¶
Overview ¶
Package memoryworkflow provides an in-process workflow.Driver for local development and single-replica deployments. It runs every run in a goroutine of the process, so runs are lost on restart and are not shared between replicas. It uses the system clock for sleeps, retries, timeouts, and delays; tests move time with testing/synctest. Definitions use workflow.New and workflow.NewStep builder callbacks; configure required workflow.Builder.MaxDuration before Handlers. Retained definition drafts cannot alter the published registry. Per-call step policy changes use workflow.Step.Derive, not driver configuration.
New keeps the registry positional and accepts Builder callbacks for its shared logger. They run directly once in order on one fresh draft. Nil entries panic; nil/empty slice expansion supplies no edits. Retained drafts cannot reconfigure the driver; registry handlers and loggers are not cloned.
Start copies encoded input before launching the run, including delayed, child, and spawned runs. Signal copies each accepted value before buffering it. Callers may reuse these byte buffers after the call returns. Each Result read and child join returns independent encoded output, so mutating a read cannot change the retained result. Copies preserve the encoded bytes; they do not clone codecs, handlers, or values captured by application functions.
Cancel and timeouts end a run by canceling its context; they cannot stop code that ignores it. Driver.Stop waits for workflow and step-attempt goroutines until its own context ends; an expired Stop cannot force them out. Attempt telemetry finishes when the handler returns, not when its supervisor stops waiting, so a late attempt can outlive the reported run outcome.
Configure worker telemetry with workflow.NewRegistry and workflow.RegistryBuilder.Telemetry, independently of the client's setting. Nil follows the current process default per actual attempt. The driver copies workflow.StartOptions.Trace, forwards it to child/spawn runs, and extracts it onto each independently cancelable step context. It never retains the start caller's context or uses application metadata as a propagation carrier.
During application composition:
registry := compose.Must(workflow.Handlers(Onboard, SendWelcome)) driver := memoryworkflow.New(registry) compose.Cleanup(driver.Stop) client := workflow.NewClient(driver)
Index ¶
- Variables
- type Builder
- type BuilderFunc
- type Driver
- func (driver *Driver) Cancel(ctx context.Context, name, id string) error
- func (driver *Driver) Result(ctx context.Context, name, id string) ([]byte, error)
- func (driver *Driver) Signal(ctx context.Context, name, id, signal string, v []byte) error
- func (driver *Driver) Start(ctx context.Context, name, id string, in []byte, o workflow.StartOptions) error
- func (driver *Driver) Status(ctx context.Context, name, id string) (workflow.Status, error)
- func (driver *Driver) Stop(ctx context.Context) error
Constants ¶
This section is empty.
Variables ¶
var ErrQueueFull = engine.ErrQueueFull
ErrQueueFull is returned by Signal when a run already buffers the maximum number of undelivered values of one signal. It does not discard the existing queue; the rejected value is not enqueued.
if errors.Is(err, memoryworkflow.ErrQueueFull) {
return fmt.Errorf("signal was not accepted: %w", err)
}
var ErrStopped = engine.ErrStopped
ErrStopped is returned by Start after Driver.Stop. A stopped instance cannot be restarted; construct another instance instead.
if errors.Is(err, memoryworkflow.ErrStopped) {
return fmt.Errorf("workflow worker is shutting down: %w", err)
}
Functions ¶
This section is empty.
Types ¶
type Builder ¶
type Builder struct {
// contains filtered or unexported fields
}
Builder edits an in-process driver's construction draft. New copies the completed logger selection, not the logger or the required registry.
driver := memoryworkflow.New(registry, func(driver *memoryworkflow.Builder) {
driver.Logger(slog.Default())
})
type BuilderFunc ¶
type BuilderFunc = func(*Builder)
BuilderFunc edits the fresh draft used by New without executing handlers. Callbacks run synchronously once in order; supplied nil callbacks panic. Later edits through a retained draft do not replace the driver's logger.
Example:
policy := memoryworkflow.BuilderFunc(func(driver *memoryworkflow.Builder) {
driver.Logger(slog.Default())
})
driver := memoryworkflow.New(registry, policy)
type Driver ¶
type Driver struct {
// contains filtered or unexported fields
}
Driver implements workflow.Driver and runs the definitions of a workflow.Registry in goroutines of the process.
It is single-process only: runs are lost on restart, and finished runs stay in memory for workflow.WorkflowOptions.Retention, one hour when zero, and are then forgotten. A run with a Delay is Running while it waits. Each signal name buffers at most 1024 values per run; Signal then returns ErrQueueFull. Call Stop at the acquiring owner's cleanup boundary, even after canceled runs: handler goroutines may still be completing cooperative cleanup.
driver := memoryworkflow.New(registry)
client := workflow.NewClient(driver)
defer func() {
if err := driver.Stop(ctx); err != nil {
logger.ErrorContext(ctx, "workflow cleanup incomplete", "error", err)
}
}()
_ = client.Driver()
func New ¶
func New(registry *workflow.Registry, configure ...BuilderFunc) *Driver
New creates a driver that runs the handlers of registry. A nil registry runs nothing. It never fails or invokes handlers during construction. Callbacks run directly once in order on one fresh draft. Nil entries panic; expanded nil/empty slices omit edits. Retaining the draft cannot change it.
driver := memoryworkflow.New(registry, func(driver *memoryworkflow.Builder) {
driver.Logger(slog.Default())
})
func (*Driver) Cancel ¶
Cancel ends the run with workflow.ErrCanceled and waits until it has stopped or ctx is done. A finished run is left as it is. Waiting here observes the terminal run record, not termination of arbitrary handlers. Stop joins workflow and step-attempt goroutines during cleanup.
if err := driver.Cancel(ctx, "onboard", "user-42"); err != nil {
return err
}
_, err := driver.Result(ctx, "onboard", "user-42")
if err != nil && !errors.Is(err, workflow.ErrCanceled) {
return err
}
func (*Driver) Result ¶
Result waits for the run to end or ctx to finish. Successful reads return independently owned encoded output; mutating it cannot change later reads. It returns workflow.ErrNotFound for an unknown or expired run. Canceling ctx ends only this wait, not the run.
output, err := driver.Result(ctx, "onboard", "user-42")
if err != nil {
return err
}
var greeting string
if err := workflow.DecodeValue(nil, output, &greeting); err != nil {
return err
}
_ = greeting
func (*Driver) Signal ¶
Signal buffers a copy of v for the run, so a signal sent before the run waits is not lost and callers may reuse v after the call. It returns workflow.ErrNotFound for an unknown or finished run and ErrQueueFull when the signal's queue is full. Copying does not validate the encoded payload.
value := []byte(`{"v":true}`)
if err := driver.Signal(ctx, "onboard", "user-42", "approved", value); err != nil {
return err
}
clear(value) // The buffered signal is unchanged.
func (*Driver) Start ¶
func (driver *Driver) Start(ctx context.Context, name, id string, in []byte, o workflow.StartOptions) error
Start begins a run in a new goroutine. It returns workflow.ErrAlreadyStarted for an existing run and workflow.ErrNotFound for an unregistered workflow. It copies in before returning, so callers may reuse the input buffer during a delayed start or execution. Copying does not validate or change the encoded payload.
input := []byte(`{"v":"welcome"}`)
if err := driver.Start(ctx, "onboard", "user-42", input, workflow.StartOptions{}); err != nil {
return err
}
clear(input) // The run keeps its own encoded input.
func (*Driver) Status ¶
Status returns the state of the run. It observes the terminal supervisor decision, not whether every handler goroutine has returned. Unknown or expired runs return workflow.ErrNotFound.
status, err := driver.Status(ctx, "onboard", "user-42")
if err != nil {
return err
}
fmt.Println(status)
func (*Driver) Stop ¶
Stop cancels every run and waits for their goroutines, or until ctx is done. Later calls to Start return ErrStopped. There is no Start method to call first: the driver is live after New.
A run is only as stoppable as its code: a workflow function or step that ignores its context keeps its goroutine after a cancel or a workflow.Builder.MaxDuration timeout, although the run itself is already reported as canceled or failed. Stop waits for such goroutines until they return or ctx ends, and in the latter case returns the error of ctx and leaves them running. Give Stop a context with a deadline. A later Stop can still join handlers that were running when an earlier cleanup deadline expired; no call can forcibly terminate application code.
cleanup, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := driver.Stop(cleanup); err != nil {
return err // Some application handlers may still be running.
}
Source Files
Directories