flow

func SkipCaller ¶

func SkipCaller(delta int) AutoNamedOption

SkipCaller adds a delta to the number of skipped stack frames.

This is useful when wrapping decorators inside wrapper functions, allowing AutoNamed to skip intermediate layers and identify the original caller.

Example:

func SomeStep() flow.Step[*Task] {
    return TaskStep(func(ctx context.Context, task *Task) error {
        // Implementation
        return nil
    })
}

func TaskStep(step flow.Step[*Task]) flow.Step[*Task] {
    // Skip TaskStep so AutoNamed picks SomeStep instead
    return AutoNamed(Retry(step), flow.SkipCaller(1))
}

func WithFullJitter ¶

func WithFullJitter() BackoffOption

WithFullJitter applies full jitter to backoff delays.

The actual delay will be a random value between 0 and the calculated delay. This is the AWS-recommended approach for maximum desynchronization and is effective at preventing thundering herd problems in distributed systems.

Cancels out any WithPercentageJitter option if both are provided. The last option in the list takes precedence.

Applies to both FixedBackoff and ExponentialBackoff.

func WithMaxDelay ¶

func WithMaxDelay(max time.Duration) BackoffOption

WithMaxDelay caps the maximum backoff delay.

This is useful for preventing exponential backoff from growing unbounded. For example, WithMaxDelay(30*time.Second) ensures retries never wait longer than 30 seconds, even if the exponential calculation would produce a larger value.

The cap is applied after jitter is calculated, ensuring the final delay (including jitter) never exceeds the maximum.

Applies to both FixedBackoff and ExponentialBackoff.

func WithMultiplier ¶

func WithMultiplier(m float64) BackoffOption

WithMultiplier sets the exponential growth rate for ExponentialBackoff.

The default multiplier is 2.0, which doubles the delay on each retry. For example, WithMultiplier(1.5) creates gentler growth (1.5×), while WithMultiplier(3.0) creates more aggressive growth (3×).

Only applies to ExponentialBackoff. This option is ignored by FixedBackoff.

func WithPercentageJitter ¶

func WithPercentageJitter(percent float64) BackoffOption

WithPercentageJitter applies percentage-based jitter to backoff delays.

For example, WithPercentageJitter(0.2) adds ±20% randomness to the delay. The actual delay will be between (delay × (1-percent)) and (delay × (1+percent)).

This provides more predictable retry timing than full jitter while still helping to desynchronize retries across multiple clients.

Cancels out any WithFullJitter option if both are provided. The last option in the list takes precedence.

Applies to both FixedBackoff and ExponentialBackoff.

func Apply ¶

func Apply[T, U any](f Consume[T, U]) Consume[T, []U]

Apply consumes each element in a slice (serial, fail-fast).

This enables clean serial iteration without orchestrator wrappers. It's fully compositional with existing Feed and Pipeline.

Example:

// Simple: extract and apply
flow.With(
    GetUsers,                             // Extract[T, []User]
    Apply(SaveUser),                      // Consume[T, User]
)                                         // Step[T]

// Pipeline: extract, transform, apply
flow.Pipeline(
    GetIDs,                               // Extract[T, []int]
    Render(LoadEntity),                   // Transform to []Entity
    Apply(ProcessEntity),                 // Consume each
)

// Composed consumption
flow.With(
    GetRecords,
    Apply(flow.Feed(
        ValidateRecord,
        SaveRecord,
    )),
)

func AutoNamedConsume ¶

func AutoNamedConsume[T, U any](consume Consume[T, U], opts ...AutoNamedOption) Consume[T, U]

AutoNamedConsume wraps a Consume with a name automatically derived from the calling function.

When tracing is enabled (via Traced), AutoNamedConsume automatically records execution events, just like NamedConsume.

See AutoNamed for further details.

func Feed ¶

func Feed[T, A, B any](
	transform Transform[T, A, B],
	consume Consume[T, B],
) Consume[T, A]

Feed composes a Transform and a Consume to produce a new Consume.

This combines the "write side" of a data pipeline: transforming an input value and feeding it to a consumer. The result is a new Consume that accepts the original input type A.

Example:

saveConfig := Feed(
    ParseConfig,  // Transform[*State, []byte, Config]
    SaveToDB,     // Consume[*State, Config]
)                 // Consume[*State, []byte]

func NamedConsume ¶

func NamedConsume[T any, U any](
	name string,
	consume Consume[T, U],
) Consume[T, U]

NamedConsume wraps a Consume with a name.

The name is prepended to the error message of the Consume, separated by a colon. For example, if the Consume returns an error "invalid config", the name is prepended to the error message: "example: invalid config".

When tracing is enabled (via Traced), NamedConsume automatically records execution events including timing and errors.

This is useful for debugging and logging.

func AutoNamedExtract ¶

func AutoNamedExtract[T, U any](extract Extract[T, U], opts ...AutoNamedOption) Extract[T, U]

AutoNamedExtract wraps an Extract with a name automatically derived from the calling function.

When tracing is enabled (via Traced), AutoNamedExtract automatically records execution events, just like NamedExtract.

See AutoNamed for further details.

func Collect ¶

func Collect[T, U any](f Extract[T, U]) Extract[T, []U]

Collect repeatedly extracts until ErrExhausted, collecting all results.

This enables pull-based iteration patterns while staying within the Extract algebra. The extractor should return ErrExhausted when no more items are available.

Example:

func NextQueueItem(ctx context.Context, state *State) (Item, error) {
    if state.queue.IsEmpty() {
        return Item{}, flow.ErrExhausted
    }
    return state.queue.Pop(), nil
}

// Collect all items from queue
allItems := Collect(NextQueueItem)  // Extract[*State, []Item]

// Use in pipeline
flow.Pipeline(
    Collect(NextQueueItem),
    Render(TransformItem),
    Apply(SaveItem),
)

func From ¶

func From[T, A, B any](
	extract Extract[T, A],
	transform Transform[T, A, B],
) Extract[T, B]

From composes an Extract and a Transform to produce a new Extract.

This combines the "read side" of a data pipeline: extracting a value from state and transforming it into another value. The result is a new Extract that performs both operations.

Example:

validatedConfig := From(
    GetRawConfig,      // Extract[*State, []byte]
    ParseAndValidate,  // Transform[*State, []byte, Config]
)                      // Extract[*State, Config]

func NamedExtract ¶

func NamedExtract[T any, U any](
	name string,
	extract Extract[T, U],
) Extract[T, U]

NamedExtract wraps an Extract with a name.

The name is prepended to the error message of the Extract, separated by a colon. For example, if the Extract returns an error "invalid config", the name is prepended to the error message: "example: invalid config".

When tracing is enabled (via Traced), NamedExtract automatically records execution events including timing and errors.

This is useful for debugging and logging.

func Traced ¶ added in v0.2.0

func Traced[T any](step Step[T], opts ...TraceOption) Extract[T, *Trace]

Traced wraps a workflow and returns an Extract that produces a Trace.

The trace records execution events for all Named steps within the workflow. This enables debugging, performance analysis, and understanding execution paths.

Events are recorded in approximate start order but may have minor variations in parallel workflows due to mutex contention. For precise chronological ordering, sort events by their Start time.

Options can be provided to configure streaming and context optimization. See WithStreamTo.

Example:

workflow := flow.Do(
    flow.Named("validate", ValidateConfig),
    flow.Named("connect", ConnectDB),
    flow.Named("migrate", RunMigrations),
)

// Basic tracing
err := flow.Spawn(
    flow.Traced(workflow),
    flow.WriteTextTo(os.Stdout),
)(ctx, state)

// Tracing with streaming to file
f, _ := os.Create("trace.jsonl")
defer f.Close()
trace, err := flow.Traced(workflow, flow.WithStreamTo(f))(ctx, state)

Tracing is opt-in and has minimal overhead when not used. All Named, NamedExtract, NamedTransform, NamedConsume, and AutoNamed variants automatically record events when a trace is present in the context.

func Value ¶ added in v0.2.0

func Value[T, U any](u U) Extract[T, U]

Value lifts a constant value into an Extract.

This creates an Extract that ignores state and context, always returning the given value. It's the monadic return/pure operation for Extract, useful for closing over values in compositions.

Example:

// Close over a slice in ForEach
flow.ForEach(
    flow.Value[*State](items),
    ProcessItem,
)

// Use in Pipeline
flow.Pipeline(
    flow.Value[*State](config),
    ValidateConfig,
    SaveConfig,
)

func And ¶

func And[T any](predicates ...Predicate[T]) Predicate[T]

And combines multiple predicates with logical AND.

All predicates must return true for the result to be true. Evaluation short-circuits on the first false or error.

Example:

When(
    And(IsProduction(), IsHealthy()),
    DeployToProduction,
)

func Not ¶

func Not[T any](predicate Predicate[T]) Predicate[T]

Not negates a predicate, returning true when the predicate returns false and vice versa.

Example:

While(Not(ServiceIsHealthy()), CheckAndWait())

func Or ¶

func Or[T any](predicates ...Predicate[T]) Predicate[T]

Or combines multiple predicates with logical OR.

Returns true if any predicate returns true. Evaluation short-circuits on the first true or error.

Example:

When(
    Or(IsDevEnvironment(), HasFeatureFlag()),
    EnableExperimentalFeature,
)

func ExponentialBackoff ¶

func ExponentialBackoff(base time.Duration, opts ...BackoffOption) RetryPredicate

ExponentialBackoff waits with exponentially increasing delays before each retry.

By default, the delay for attempt N is base × 2^(N-1). For example, with a base of 100ms:

• Attempt 1: 100ms
• Attempt 2: 200ms
• Attempt 3: 400ms

If the number of attempts is less than 1, or the calculated delay overflows, the base delay is used instead.

If the context is cancelled during the wait, the predicate returns false.

Options:

• WithFullJitter randomizes delay between 0 and the calculated delay
• WithPercentageJitter adds ±N% randomness to the calculated delay
• WithMaxDelay caps the maximum delay
• WithMultiplier changes the growth rate (default 2.0)

func FixedBackoff ¶

func FixedBackoff(delay time.Duration, opts ...BackoffOption) RetryPredicate

FixedBackoff waits for a fixed duration before each retry.

The delay is applied before each retry. If the context is cancelled during the wait, the predicate returns false and the retry is aborted.

Options:

• WithFullJitter randomizes delay between 0 and the fixed duration
• WithPercentageJitter adds ±N% randomness to the fixed duration
• WithMaxDelay caps the delay (useful with jitter)
• WithMultiplier ignored (only applies to ExponentialBackoff)

func OnlyIf ¶

func OnlyIf(check func(error) bool) RetryPredicate

OnlyIf conditionally retries based on the error.

The predicate returns true only if the provided check function returns true for the error. This is useful for retrying only transient errors while immediately failing on permanent errors like validation failures.

func UpTo ¶

func UpTo(maxAttempts int) RetryPredicate

UpTo limits retries to a maximum number of attempts.

The predicate returns true if attempts < maxAttempts, allowing retries to continue until the limit is reached.

func AutoNamed ¶

func AutoNamed[T any](step Step[T], opts ...AutoNamedOption) Step[T]

AutoNamed wraps a Step with a name automatically derived from the calling function.

This uses reflection to extract the name of the function that calls AutoNamed, which is useful for reducing repetition when naming steps in step constructors.

When tracing is enabled (via Traced), AutoNamed steps automatically record execution events, just like Named steps.

Example:

func CreateDatabase() flow.Step[*Config] {
    return flow.AutoNamed(func(ctx context.Context, cfg *Config) error {
        // Implementation
        return nil
    })
}
// If this step fails, the error will be prefixed with "CreateDatabase: ..."
// When traced, the event will be named "CreateDatabase"

Note: AutoNamed only works when called directly from a named function. It will not work correctly when called from anonymous functions or closures.

func Do ¶

func Do[T any](steps ...Step[T]) Step[T]

Do executes multiple steps in order, one at a time.

This is the simplest way to run a sequence of static steps. For dynamic step sequences (StepsProvider), use InSerial instead.

Do is the same as DoWith with the default Options.

Example:

Do(
    ValidateConfig(),
    StartDatabase(),
    StartServer(),
)

func DoWith ¶

func DoWith[T any](opts Options, steps ...Step[T]) Step[T]

DoWith executes multiple steps in order, one at a time, with custom options.

This allows configuring behavior like error joining. Use Do for the common case with default options.

Example:

// Run all cleanup steps even if some fail, collect all errors
DoWith(
    Options{JoinErrors: true},
    CleanupTempFiles(),
    CloseConnections(),
    RemoveLockFiles(),
)

func IgnoreError ¶

func IgnoreError[T any](step Step[T]) Step[T]

IgnoreError wraps a step to always return nil, even if the step fails.

This is useful for "best effort" operations where failures should not stop the overall flow.

func InParallel ¶

func InParallel[T any](providers ...StepsProvider[T]) Step[T]

InParallel combines multiple step sequences and runs them concurrently.

InParallel is the same as InParallelWith with the default ParallelOptions.

func InParallelWith ¶

func InParallelWith[T any](
	opts ParallelOptions,
	providers ...StepsProvider[T],
) Step[T]

InParallelWith combines multiple step sequences and runs them concurrently.

All step sequences are expanded sequentially first, then the resulting steps are run in separate goroutines.

func InSerial ¶

func InSerial[T any](providers ...StepsProvider[T]) Step[T]

InSerial combines multiple step sequences and runs them in order, one at a time.

InSerial is the same as InSerialWith with the default Options.

Example:

InSerial(
    flow.ForEach(GetFoos, ProcessFoo),
    flow.ForEach(GetBars, ProcessBar),
)

func InSerialWith ¶

func InSerialWith[T any](
	opts Options,
	providers ...StepsProvider[T],
) Step[T]

InSerialWith combines multiple step sequences and runs them in order, one at a time.

func Named ¶

func Named[T any](name string, step Step[T]) Step[T]

Named wraps a Step with a name.

The name is prepended to the error message of the Step, separated by a colon. For example, if the Step returns an error "invalid config", the name is prepended to the error message: "example: invalid config".

Named also maintains a stack of step names in the context, which can be retrieved using Names. When Named decorators are nested, each appends its name to the stack, creating a hierarchical path (e.g., "process.parse.validate").

When tracing is enabled (via Traced), Named automatically records execution events including timing and errors.

This is useful for debugging and logging.

func OnError ¶

func OnError[T any](step Step[T], onError Transform[T, error, Step[T]]) Step[T]

OnError provides dynamic error handling with fallback steps.

If the primary step fails, onError is called with the error and can return a fallback step to execute instead. The onError transform has access to the context, state, and the error, allowing for sophisticated error handling strategies.

If onError itself returns an error, that error is propagated and no fallback step is executed.

Example:

OnError(
    CallPrimaryAPI,
    func(ctx context.Context, state *State, err error) (Step[*State], error) {
        if errors.Is(err, ErrTimeout) {
            return CallBackupAPI, nil
        }
        if errors.Is(err, ErrRateLimit) {
            return RetryAfterDelay, nil
        }
        return nil, fmt.Errorf("unrecoverable: %w", err)
    },
)

func Pipeline ¶

func Pipeline[T, A, B any](
	extract Extract[T, A],
	transform Transform[T, A, B],
	consume Consume[T, B],
) Step[T]

Pipeline composes an Extract, Transform, and Consume into a complete Step.

This creates a full data pipeline: extract a value from state, transform it, and consume the result. This is a common pattern for workflows that follow the read-process-write model.

Pipeline is equivalent to:

With(From(extract, transform), consume)
With(extract, Feed(transform, consume))

Example:

processConfig := Pipeline(
    GetRawConfig,      // Extract[*State, []byte]
    ParseAndValidate,  // Transform[*State, []byte, Config]
    SaveToDatabase,    // Consume[*State, Config]
)                      // Step[*State]

func RecoverPanics ¶

func RecoverPanics[T any](step Step[T]) Step[T]

RecoverPanics wraps a step to recover from panics and convert them to errors.

If the step panics, the panic value is wrapped in a RecoveredPanic error. This is useful for defensive programming when calling code that may panic.

func Retry ¶

func Retry[T any](
	step Step[T],
	predicates ...RetryPredicate,
) Step[T]

Retry executes a step and retries it on failure based on the given predicates.

All predicates must return true for a retry to occur. If any predicate returns false, the last error is returned immediately.

If no predicates are provided, this defaults to retrying up to 3 times with exponential backoff starting at 100ms and full jitter to prevent thundering herd problems.

func Sleep ¶

func Sleep[T any](duration time.Duration) Step[T]

Sleep pauses execution for the specified duration.

The sleep respects context cancellation, returning context.Canceled if the context is cancelled before the duration elapses.

This is useful in polling loops or when adding delays between operations.

Example:

flow.While(
    ServiceNotReady(),
    flow.Do(
        CheckStatus(),
        flow.Sleep(5*time.Second),
    ),
)

func Spawn ¶

func Spawn[Parent, Child any](
	derive Extract[Parent, Child],
	step Step[Child],
) Step[Parent]

Spawn executes a step with a different state type derived from the parent.

This is useful when you need to run a workflow in a narrower context. The derive function extracts or constructs child state from the parent, then the step runs with that child state.

Example:

// Run database setup with DB-specific state
Spawn(
    PrepareDbConnection,    // Extract[*EnvSetup, *DbSetup]
    RunDbMigrations,        // Step[*DbSetup]
)                           // Step[*EnvSetup]

func Unless ¶

func Unless[T any](
	predicate Predicate[T],
	step Step[T],
) Step[T]

Unless runs the given step only if the predicate returns false.

If the predicate returns true, the step is skipped and nil is returned. If the predicate returns an error, that error is propagated.

func When ¶

func When[T any](
	predicate Predicate[T],
	step Step[T],
) Step[T]

When runs the given step only if the predicate returns true.

If the predicate returns false, the step is skipped and nil is returned. If the predicate returns an error, that error is propagated.

func While ¶

func While[T any](
	predicate Predicate[T],
	step Step[T],
) Step[T]

While repeatedly executes the step as long as the predicate returns true.

The predicate is evaluated before each iteration. If it returns false or an error, the loop terminates. If the step returns an error, that error is propagated immediately.

Example:

While(
    ServiceIsNotReady(),
    CheckStatus(),
)

To prevent infinite loops, combine with WithTimeout or use predicates that eventually become false.

func With ¶

func With[T, U any](
	extract Extract[T, U],
	consume Consume[T, U],
) Step[T]

With combines an Extract and a Consume into a Step.

This creates a two-stage workflow: extract a value from state, then process/consume it. This is useful when you need to read some data and perform an action with it.

Example:

processUsers := With(
    GetUsers,      // Extract[*State, []User]
    SendEmails,    // Consume[*State, []User]
)                  // Step[*State]

func WithDeadline ¶

func WithDeadline[T any](deadline time.Time, step Step[T]) Step[T]

WithDeadline wraps a step with an absolute deadline.

The step is executed with a derived context that will be cancelled at the specified time. If the step does not complete before the deadline, it will receive a cancelled context and should return context.DeadlineExceeded.

Example:

deadline := time.Now().Add(5 * time.Second)
flow.WithDeadline(deadline, ExpensiveOperation())

func WithLogger ¶

func WithLogger[T any](logger *log.Logger, step Step[T]) Step[T]

WithLogger configures a Step to use a specific log.Logger for logging.

The logger is stored in the context and used by WithLogging. If no logger is configured, WithLogging will use log.Default.

This is typically applied once at the root of a workflow to configure logging for all nested steps. For optimal performance, use Named or AutoNamed liberally throughout your workflow—the flow library consolidates all context state into a single lookup via flowCtx, but this optimization only provides constant-time access when step names are present in the context chain. Using WithLogger without corresponding Named steps may incur linear context chain traversal cost.

Example:

myLogger := log.New(os.Stdout, "workflow: ", log.LstdFlags)
step := WithLogger(myLogger,
    Named("process",
        WithLogging(processStep)))

func WithLogging ¶

func WithLogging[T any](step Step[T]) Step[T]

WithLogging wraps a Step with logging that prints messages when the step starts and finishes, including execution duration.

The log messages include the full dotted path of step names from the context (e.g., "process.parse.validate"). If no names are set, logs show "<unknown>".

The logger used is retrieved from the context (set by WithLogger). If no logger is configured, log.Default is used.

Log format:

[step.name] starting step
[step.name] finished step (took 123ms)

Example:

step := Named("process",
    WithLogging(
        Named("parse",
            WithLogging(parseStep))))

This would log:

[process] starting step
[process.parse] starting step
[process.parse] finished step (took 5ms)
[process] finished step (took 10ms)

For custom logging formats, use Names to retrieve the name stack and implement your own logging decorator.

func WithSlogger ¶

func WithSlogger[T any](logger *slog.Logger, step Step[T]) Step[T]

WithSlogger configures a Step to use a specific slog.Logger for structured logging.

The logger is stored in the context and used by WithSlogging. If no logger is configured, WithSlogging will use slog.Default.

This is typically applied once at the root of a workflow to configure logging for all nested steps. For optimal performance, use Named or AutoNamed liberally throughout your workflow—the flow library consolidates all context state into a single lookup via flowCtx, but this optimization only provides constant-time access when step names are present in the context chain. Using WithSlogger without corresponding Named steps may incur linear context chain traversal cost.

Example:

myLogger := slog.New(slog.NewJSONHandler(os.Stdout, nil))
step := WithSlogger(myLogger,
    Named("process",
        WithSlogging(slog.LevelInfo, processStep)))

func WithSlogging ¶

func WithSlogging[T any](level slog.Level, step Step[T]) Step[T]

WithSlogging wraps a Step with structured logging that emits log records when the step starts and finishes, including execution duration.

The log records include the full dotted path of step names from the context as a "name" attribute (e.g., "process.parse.validate"). If no names are set, the name attribute will be "<unknown>". The finish log includes a "duration_ms" attribute with the execution time in milliseconds.

The logger used is retrieved from the context (set by WithSlogger). If no logger is configured, slog.Default is used.

Example:

step := Named("process",
    WithSlogging(slog.LevelInfo,
        Named("parse",
            WithSlogging(slog.LevelDebug, parseStep))))

This would emit structured log records similar to:

{"level":"info","msg":"starting step","name":"process"}
{"level":"debug","msg":"starting step","name":"process.parse"}
{"level":"debug","msg":"finished step","name":"process.parse","duration_ms":5}
{"level":"info","msg":"finished step","name":"process","duration_ms":10}

For custom logging formats, use Names to retrieve the name stack and implement your own logging decorator.

func WithTimeout ¶

func WithTimeout[T any](timeout time.Duration, step Step[T]) Step[T]

WithTimeout wraps a step with a timeout.

The step is executed with a derived context that will be cancelled after the specified duration. If the step does not complete within the timeout, it will receive a cancelled context and should return context.DeadlineExceeded.

Example:

flow.WithTimeout(5*time.Second, ExpensiveOperation())

func WriteFlatTextTo ¶ added in v0.2.0

func WriteFlatTextTo(w io.Writer) Step[*Trace]

WriteFlatTextTo returns a Step that serializes a trace to flat text format.

This enables natural composition with Spawn. The flat format shows events in chronological order without tree indentation, making it more suitable for analyzing parallel workflows.

workflow := flow.Spawn(
    flow.Traced(myWorkflow),
    flow.WriteFlatTextTo(os.Stdout),
)

func WriteJSONTo ¶ added in v0.2.0

func WriteJSONTo(w io.Writer) Step[*Trace]

WriteJSONTo returns a Step that serializes a trace to JSON.

This enables natural composition with Spawn:

workflow := flow.Spawn(
    flow.Traced(myWorkflow),
    flow.WriteJSONTo(os.Stdout),
)

func WriteTextTo ¶ added in v0.2.0

func WriteTextTo(w io.Writer) Step[*Trace]

WriteTextTo returns a Step that serializes a trace to human-readable text.

This enables natural composition with Spawn:

workflow := flow.Spawn(
    flow.Traced(myWorkflow),
    flow.WriteTextTo(os.Stdout),
)

func ForEach ¶

func ForEach[T, U any](
	extract Extract[T, []U],
	f func(U) Step[T],
) StepsProvider[T]

ForEach is a convenience helper that combines Extract and Map.

It extracts a slice and maps each element to a step, resulting in a StepsProvider. This is useful when you want the flexibility to choose serial or parallel execution at the orchestration level.

Example:

// Define once
processWorkflow := flow.ForEach(GetItems, ProcessItem)

// Use serially
flow.InSerial(processWorkflow)

// Use in parallel
flow.InParallel(processWorkflow)

func Steps ¶

func Steps[T any](steps ...Step[T]) StepsProvider[T]

Steps creates a static step sequence from the provided steps.

This is useful when you need to convert static steps into a StepsProvider for use with InSerial or InParallel. For simple static sequences, prefer using Do directly.

Example:

// Prefer Do for simple static sequences
Do(ValidateConfig(), StartDatabase(), RunMigrations())

// Use Steps when you need a StepsProvider
InSerial(
    Steps(ValidateConfig(), StartDatabase()),
    flow.ForEach(GetServices, DeployService),
)

func DepthAtMost ¶ added in v0.2.0

func DepthAtMost(depth int) TraceFilter

DepthAtMost returns a filter that matches events at or above the given depth.

For example, DepthAtMost(2) matches steps at depth 1 or 2.

func DepthEquals ¶ added in v0.2.0

func DepthEquals(depth int) TraceFilter

DepthEquals returns a filter that matches events at the given depth.

Depth is defined as len(Names). For example:

• depth 1: top-level steps
• depth 2: first level of nesting
• depth 3: second level of nesting

func ErrorMatches ¶ added in v0.2.0

func ErrorMatches(pattern string) TraceFilter

ErrorMatches returns a filter that matches events where the error message matches the glob pattern.

If the pattern is malformed, no events will match (returns false).

Example patterns:

• "*timeout*" matches errors containing "timeout"
• "connection *" matches errors starting with "connection "
• "database error" matches exactly "database error"

func HasError ¶ added in v0.2.0

func HasError() TraceFilter

HasError returns a filter that matches events with errors.

func HasPathPrefix ¶ added in v0.2.0

func HasPathPrefix(prefix []string) TraceFilter

HasPathPrefix returns a filter that matches events where the full Names path has the given prefix.

Example:

// Match all steps under ["migrate", "tables"]
filter := flow.HasPathPrefix([]string{"migrate", "tables"})

func MaxDuration ¶ added in v0.2.0

func MaxDuration(d time.Duration) TraceFilter

MaxDuration returns a filter that matches events with duration <= d.

func MinDuration ¶ added in v0.2.0

func MinDuration(d time.Duration) TraceFilter

MinDuration returns a filter that matches events with duration >= d.

func NameMatches ¶ added in v0.2.0

func NameMatches(pattern string) TraceFilter

NameMatches returns a filter that matches events where the step name (last element of Names) matches the glob pattern.

Patterns use filepath.Match semantics—see its documentation for the full details of matching behavior (e.g., *, ?, and [...] for character sets).

If the pattern is malformed, no events will match (returns false).

func NamePrefix ¶ added in v0.2.0

func NamePrefix(prefix string) TraceFilter

NamePrefix returns a filter that matches events where the step name (last element of Names) has the given prefix.

func NoError ¶ added in v0.2.0

func NoError() TraceFilter

NoError returns a filter that matches events without errors.

func PathMatches ¶ added in v0.2.0

func PathMatches(pattern string) TraceFilter

PathMatches returns a filter that matches events where the full dotted path matches the glob pattern.

Patterns use filepath.Match semantics—see its documentation for the full details of matching behavior. The path is constructed by joining Names with "." separators.

If the pattern is malformed, no events will match (returns false).

func TimeRange ¶ added in v0.2.0

func TimeRange(start, end time.Time) TraceFilter

TimeRange returns a filter that matches events that started within the given time range.

Both start and end times are inclusive. Events are matched if their Start time falls between start and end.

Example:

// Find events that started in the first second
workflowStart := trace.Events()[0].Start
filter := flow.TimeRange(workflowStart, workflowStart.Add(time.Second))

func WithStreamTo ¶ added in v0.2.0

func WithStreamTo(w io.Writer) TraceOption

WithStreamTo configures the trace to stream events as JSON Lines to the given writer.

Events are written in JSON Lines format (one event per line) as they complete, enabling real-time monitoring and ensuring traces are preserved even if the process crashes. This is different from WriteTo/WriteJSONTo which output a pretty-printed JSON array after execution completes.

All events are retained in memory for post-execution querying.

Write failures to the stream are best-effort and do not cause the workflow to fail. This ensures that tracing infrastructure never breaks the workflow itself.

Example:

f, _ := os.Create("trace.jsonl")
defer f.Close()
trace, err := flow.Traced(workflow, flow.WithStreamTo(f))(ctx, state)

// trace.jsonl contains one JSON object per line:
// {"step_names":["validate"],"start":"...","duration":45000000}
// {"step_names":["connect"],"start":"...","duration":120000000}
// ...

func AutoNamedTransform ¶

func AutoNamedTransform[T, A, B any](transform Transform[T, A, B], opts ...AutoNamedOption) Transform[T, A, B]

AutoNamedTransform wraps a Transform with a name automatically derived from the calling function.

When tracing is enabled (via Traced), AutoNamedTransform automatically records execution events, just like NamedTransform.

See AutoNamed for further details.

func Chain ¶

func Chain[T, A, B, C any](
	first Transform[T, A, B],
	second Transform[T, B, C],
) Transform[T, A, C]

Chain composes two Transforms into a single Transform.

This creates a transformation pipeline where the output of the first transform becomes the input to the second. Both transforms have access to the same state T.

Example:

parseAndValidate := Chain(
    ParseJSON,    // Transform[*State, []byte, Config]
    ValidateConfig, // Transform[*State, Config, Config]
)                 // Transform[*State, []byte, Config]

func Chain3 ¶

func Chain3[T, A, B, C, D any](
	first Transform[T, A, B],
	second Transform[T, B, C],
	third Transform[T, C, D],
) Transform[T, A, D]

Chain3 composes three Transforms into a single Transform.

This extends Chain to handle three-stage transformation pipelines. Each transform has access to the same state T.

Example:

processData := Chain3(
    Parse,      // Transform[*State, []byte, RawData]
    Validate,   // Transform[*State, RawData, ValidData]
    Normalize,  // Transform[*State, ValidData, NormData]
)               // Transform[*State, []byte, NormData]

func Chain4 ¶

func Chain4[T, A, B, C, D, E any](
	first Transform[T, A, B],
	second Transform[T, B, C],
	third Transform[T, C, D],
	fourth Transform[T, D, E],
) Transform[T, A, E]

Chain4 composes four Transforms into a single Transform.

This extends Chain to handle four-stage transformation pipelines. Each transform has access to the same state T.

For longer pipelines, you can nest Chain functions:

Chain(
    Chain4(t1, t2, t3, t4),
    Chain4(t5, t6, t7, t8),
)

Example:

processRequest := Chain4(
    Decode,     // Transform[*State, []byte, Request]
    Validate,   // Transform[*State, Request, ValidReq]
    Enrich,     // Transform[*State, ValidReq, EnrichedReq]
    Sanitize,   // Transform[*State, EnrichedReq, SafeReq]
)               // Transform[*State, []byte, SafeReq]

func FallbackTo ¶

func FallbackTo[T any](fallback Step[T]) Transform[T, error, Step[T]]

FallbackTo returns an error transform that always returns the provided fallback step.

This is useful with OnError when you unconditionally want to run a fallback step on error, regardless of the state or the original error.

func Map ¶

func Map[T any, A any](
	f func(A) Step[T],
) Transform[T, []A, []Step[T]]

Map lifts a function from A to Step[T] into a Transform from []A to []Step[T].

Map is designed to compose with From for dynamic workflows:

flow.InSerial(
    flow.From(
        GetServices,                 // Extract[*Config, []string]
        flow.Map(DeployService), // Transform[*Config, []string, []Step[*Config]]
    ),                               // → StepsProvider[*Config]
)

The call to From results in a StepsProvider, the same type used by InSerial and InParallel. This allows one to switch such a loop from serial to concurrent processing simply by changing the function that wraps From.

Map can also be used with Chain to build more complex transformations:

flow.From(
    GetRawData,                      // Extract[*State, Data]
    flow.Chain(
        ParseItems,                  // Transform[*State, Data, []Item]
        flow.Map(ProcessItem),   // Transform[*State, []Item, []Step[*State]]
    ),
)

func NamedTransform ¶

func NamedTransform[T any, In any, Out any](
	name string,
	transform Transform[T, In, Out],
) Transform[T, In, Out]

NamedTransform wraps a Transform with a name.

The name is prepended to the error message of the Transform, separated by a colon. For example, if the Transform returns an error "invalid config", the name is prepended to the error message: "example: invalid config".

When tracing is enabled (via Traced), NamedTransform automatically records execution events including timing and errors.

This is useful for debugging and logging.

func Render ¶

func Render[T, In, Out any](f Transform[T, In, Out]) Transform[T, []In, []Out]

Render transforms each element in a slice (serial, fail-fast).

This completes the Transform algebra with collection support. It enables natural composition in From and Chain for processing slices.

Example:

// Multi-stage rendering pipeline
flow.Pipeline(
    GetRawData,                           // Extract[T, [][]byte]
    flow.Chain(
        Render(Parse),                    // []byte → []Record
        Render(Validate),                 // []Record → []Record
        Render(Enrich),                   // []Record → []EnrichedRecord
    ),
    Apply(Save),
)

// Simple transformation
flow.From(
    GetUserIDs,                           // Extract[T, []int]
    Render(LoadUser),                     // Transform[T, int, User]
)                                         // Extract[T, []User]