fluxo

package module
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Dec 12, 2025 License: MIT Imports: 11 Imported by: 0

README

Fluxo — Lightweight Workflow Engine for Go

Coverage

Go Reference Go Report Card Tests

Fluxo is a fast, embeddable, deterministic workflow engine written in pure Go. It is designed as a practical alternative to Temporal/Camunda for teams that want workflow reliability without running workflow infrastructure.

Fluxo runs inside your Go service, supports multiple persistence backends, and uses a simple, ergonomic API.

✨ Features (MVP)

  • Deterministic, retryable workflow execution

  • Pluggable persistence

    • In-memory (testing/dev)
    • SQLite
    • PostgreSQL
    • Redis
    • MongoDB

  • Built-in asynchronous worker

  • Strongly-typed workflow support (via TypedStep, TypedLoop, TypedWhile)

  • Parallel, conditional, and looping control flow

  • Timers & signals

  • LocalRunner for in-process testing (non-durable)

Fluxo is a library — not a service. You embed it directly into your application.

📦 Installation

go get github.com/petrijr/fluxo

Go 1.21+ is recommended.

🚀 Quick Start

Define a workflow using the builder API and run it using an engine:

package main

import (
	"context"
	"log"
	"github.com/petrijr/fluxo"
)

func createAccount(ctx context.Context, input any) (any, error) {
	return map[string]any{"userID": "123"}, nil
}

func sendWelcomeEmail(ctx context.Context, input any) (any, error) {
	state := input.(map[string]any)
	log.Printf("sending welcome email to %s", state["userID"])
	return state, nil
}

func main() {
	ctx := context.Background()

	flow := fluxo.New("OnboardUser").
		Step("createAccount", createAccount).
		Step("sendWelcomeEmail", sendWelcomeEmail)

	eng := fluxo.NewInMemoryEngine()

	if err := flow.Register(eng); err != nil {
		log.Fatal(err)
	}

	inst, err := fluxo.Run(ctx, eng, flow.Name(), nil)
	if err != nil {
		log.Fatal(err)
	}

	log.Printf("workflow completed: id=%s status=%s", inst.ID, inst.Status)
}

🗄 Persistence Backends

Fluxo supports multiple backends. Definitions are always in-memory; instances and execution history depend on your backend choice.

In-Memory

Use for tests or ephemeral/local execution:

eng := fluxo.NewInMemoryEngine()
SQLite

Embedded durability; ideal default for single-node services:

db, _ := sql.Open("sqlite", "file:fluxo.db?_journal=WAL")
eng, _ := fluxo.NewSQLiteEngine(db)
PostgreSQL
db, _ := sql.Open("pgx", "postgres://user:pass@localhost:5432/fluxo")
eng, _ := fluxo.NewPostgresEngine(db)
Redis
rdb := redis.NewClient(&redis.Options{ Addr: "localhost:6379" })
eng := fluxo.NewRedisEngine(rdb)
MongoDB
client, _ := mongo.Connect(ctx, options.Client().ApplyURI("mongodb://localhost:27017"))
eng := fluxo.NewMongoEngine(client)

Backend choice does not change workflow code.

🔧 Control Flow

Fluxo provides simple, composable workflow primitives.

Sequential Steps
flow.Step("a", stepA).Step("b", stepB)
Conditionals
flow.If("check-limit",
func (input any) bool { return input.(int) < 100 },
fluxo.StepFunc(func (ctx context.Context, in any) (any, error) { return "ok", nil }),
fluxo.StepFunc(func (ctx context.Context, in any) (any, error) { return "too large", nil }),
)
Parallel Work
flow.Parallel("prepare",
stepFetchUser,
stepFetchSettings,
stepWarmCache,
)
Loops

Fixed-count:

flow.Loop("repeat", 3, body)

While-condition:

flow.While("until-ready",
func (input any) bool { return !input.(State).Ready },
body,
)
Typed Helpers

Avoid any by using strongly-typed steps:

flow.Step("typed", fluxo.TypedStep(func(ctx context.Context, s Counter) (Counter, error) {
s.Value++
return s, nil
}))

Typed looping:

flow.Step("loop", fluxo.TypedWhile(
func(s Counter) bool { return s.Value < 5 },
func (ctx context.Context, s Counter) (Counter, error) {
s.Value++
return s, nil
},
))

🧵 Workers (Asynchronous Execution)

Fluxo workers pull tasks from the task queue and execute them:

w := fluxo.NewWorker(eng, queue)
go w.Run(ctx)

Workers can be horizontally scaled.

🧪 LocalRunner (In-Process Testing)

LocalRunner bundles engine + queue + worker for easy test setups.

runner := fluxo.NewLocalRunner()
runner.StartWorkers(ctx, 1)
runner.StartWorkflowAsync(ctx, "MyFlow", input)

⚠️ Not crash-durable — for tests & dev only.

📊 Observability

Fluxo exposes an Observer interface for logging and metrics.

Logging
obs := fluxo.NewLoggingObserver(nil) // uses slog.Default()
eng := fluxo.NewInMemoryEngineWithObserver(obs)
Metrics
metrics := &fluxo.BasicMetrics{}
eng := fluxo.NewSQLiteEngineWithObserver(db, metrics)
snapshot := metrics.Snapshot()

⚙️ Performance

Fluxo targets < 1ms overhead per step on typical hardware (excluding user logic). This is enforced via a performance regression test in the repository.

Actual performance varies with backend choice.

🧱 Guarantees & Limitations (Honest MVP)

✔ Engine Guarantees
  • Deterministic workflow planning
  • At-least-once step execution
  • Durable workflow state when using persistent backends
  • Worker crash recovery (persistent backends only)
✔ Non-Guarantees (Current MVP)
  • No global saga/compensation framework
  • No distributed transaction guarantees
  • No cross-workflow coordination primitives
  • No built-in admin UI or orchestration service
  • LocalRunner is not durable and cannot recover from process crashes

🗺 Roadmap (Post-MVP)

These are intentionally not implemented yet but may come next:

  • Saga helpers (compensation patterns)
  • Better observability integrations (Prometheus, OpenTelemetry)
  • Workflow versioning helpers
  • CLI tooling for inspecting workflows
  • Kafka/NATS queue backends
  • More ergonomic DSL for workflow definitions
  • Optional workflow visualization tooling

🤝 Contributing

Issues, PRs, and feedback are welcome! This project is still evolving and contributions are encouraged.

📄 License

MIT — see LICENSE.

📘 API Reference (MVP)

This is the public API surface area for Fluxo’s MVP release.

Top-Level Constructors
func New(name string) *FlowBuilder
func Run(ctx context.Context, eng Engine, workflow string, input any) (*Instance, error)
Engines
func NewInMemoryEngine() Engine
func NewInMemoryEngineWithObserver(o Observer) Engine

func NewSQLiteEngine(db *sql.DB) (Engine, error)
func NewSQLiteEngineWithObserver(db *sql.DB, o Observer) (Engine, error)

func NewPostgresEngine(db *sql.DB) (Engine, error)
func NewPostgresEngineWithObserver(db *sql.DB, o Observer) (Engine, error)

func NewRedisEngine(client *redis.Client) Engine
func NewRedisEngineWithObserver(client *redis.Client, o Observer) Engine

func NewMongoEngine(client *mongo.Client) Engine
func NewMongoEngineWithObserver(client *mongo.Client, o Observer) Engine
Task Queues
func NewInMemoryQueue(capacity int) TaskQueue
func NewSQLiteQueue(db *sql.DB) (TaskQueue, error)
func NewPostgresQueue(db *sql.DB) (TaskQueue, error)
func NewRedisQueue(client *redis.Client) TaskQueue
func NewMongoQueue(client *mongo.Client) TaskQueue
Worker
func NewWorker(eng Engine, q TaskQueue) *Worker
func NewWorkerWithConfig(eng Engine, q TaskQueue, cfg worker.Config) *Worker

Key methods:

func (w *Worker) Run(ctx context.Context) error
func (w *Worker) ProcessOne(ctx context.Context) (bool, error)
LocalRunner
type LocalRunner struct {
Engine Engine
Queue  TaskQueue
Worker *Worker
}

func NewLocalRunner() *LocalRunner
func (r *LocalRunner) StartWorkers(ctx context.Context, n int) error
func (r *LocalRunner) StartWorkflowAsync(ctx context.Context, name string, input any) error
func (r *LocalRunner) SignalAsync(ctx context.Context, instanceID string, signal string, payload any) error
Observability
type Observer interface {
// lifecycle + metrics events
}

func NewLoggingObserver(logger *slog.Logger) Observer
func NewCompositeObserver(obs ...Observer) Observer
func NewNoopObserver() Observer

type BasicMetrics struct { /* counters */ }
func (m *BasicMetrics) Snapshot() BasicMetricsSnapshot
Builder API
type FlowBuilder struct {
// ...
}

func (b *FlowBuilder) Step(name string, fn StepFunc) *FlowBuilder
func (b *FlowBuilder) If(name string, cond ConditionFunc, then StepFunc, els StepFunc) *FlowBuilder
func (b *FlowBuilder) Parallel(name string, steps ...StepFunc) *FlowBuilder
func (b *FlowBuilder) Loop(name string, times int, body StepFunc) *FlowBuilder
func (b *FlowBuilder) While(name string, cond ConditionFunc, body StepFunc) *FlowBuilder
func (b *FlowBuilder) WaitForSignal(name, signal string) *FlowBuilder
func (b *FlowBuilder) Sleep(name string, dur time.Duration) *FlowBuilder
func (b *FlowBuilder) SleepUntil(name string, t time.Time) *FlowBuilder
Step Types
type StepFunc func (ctx context.Context, input any) (any, error)
type ConditionFunc func(input any) bool
Typed Helpers
func TypedStep[I, O any](fn func(context.Context, I) (O, error)) StepFunc
func TypedWhile[I any](cond func (I) bool, body func (context.Context, I) (I, error)) StepFunc
func TypedLoop[I any](times int, body func (context.Context, I) (I, error)) StepFunc

Documentation

Overview

Package fluxo provides a lightweight, embeddable workflow engine for Go.

Fluxo is designed for backend services that need reliable asynchronous operations, background tasks, or long-lived workflows—without introducing external dependencies or heavy infrastructure. It runs fully in Go, supports multiple persistence backends, and integrates cleanly into existing codebases.

Core Concepts

The Fluxo programming model is intentionally small and idiomatic:

  1. Engine
  2. Worker
  3. FlowBuilder
  4. StepFunc
  5. LocalRunner

These components form a complete workflow system with deterministic execution, durable state (when using persistent backends), and a clear mental model.

Engine

The Engine stores workflow definitions, persists workflow state, manages execution plans, and provides APIs to:

  • start workflows
  • resume workflows after steps complete
  • deliver signals
  • read workflow state and history

Engines can be backed by different storage systems:

  • In-memory (non-durable, best for tests)
  • SQLite (embedded durability)
  • Postgres
  • Redis
  • MongoDB

Each backend includes a matching task queue implementation so workers can reliably fetch work.

Engines are safe for use from background workers or from application code that wants to schedule workflows synchronously.

Worker

A Worker pulls tasks from a configured queue and executes workflow steps. Workers run asynchronously and can be scaled horizontally.

Responsibilities include:

  • polling task queues
  • executing StepFuncs deterministically
  • applying retry policies
  • driving workflows forward to completion

Applications typically run one or more workers as background goroutines or as separate processes.

FlowBuilder

FlowBuilder provides the ergonomic, declarative API used to define workflows. It supports common control-flow structures:

  • Sequential steps
  • Conditionals (If / Switch)
  • Parallel execution (Parallel / ParallelMap)
  • Loops (Loop / While, including typed variants)
  • Timers and sleeps
  • Signals

Example: 

fluxo.New("Example").
    Step("a", doA).
    Step("b", doB).
    Parallel("c",
        fluxo.StepFunc("p1", work1),
        fluxo.StepFunc("p2", work2),
    )

Definitions created with FlowBuilder are registered into an Engine before use.

StepFunc

A StepFunc is the fundamental executable unit of a workflow: 

type StepFunc func(ctx context.Context, state *State) error

Steps are:

  • deterministic: same inputs → same observable behavior
  • idempotent: may be retried if a worker crashes
  • isolated: they receive a state object representing workflow data

Typed helpers make it easy to work with structured Go values without manual marshaling.

LocalRunner

LocalRunner bundles an in-memory engine, queue, and worker into a single, process-local helper useful for development and unit testing. It lets you:

  • start workflows synchronously or asynchronously
  • send signals
  • wait for completion

LocalRunner is intentionally **not crash-durable**, but it provides the most convenient way to run and debug workflows during development.

Summary

Fluxo’s goal is to give Go developers a workflow engine that feels like Go: easy to embed, easy to test, deterministic, and without operational overhead. Engines manage workflow state, Workers execute steps, FlowBuilder defines workflows, StepFuncs contain business logic, and LocalRunner provides a fast, developer-friendly runtime.

For examples, see the /examples directory or the project README.

local_runner.go

Example (FlowBuilder)

Example_flowBuilder demonstrates defining and running a simple workflow using the high-level FlowBuilder API and an in-memory engine. 

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/petrijr/fluxo"
)

func main() {
	ctx := context.Background()

	flow := fluxo.New("Greeting").
		Step("sayHello", sayHello).
		Step("decorateMessage", decorateMessage)

	eng := fluxo.NewInMemoryEngine()

	if err := flow.Register(eng); err != nil {
		log.Fatal(err)
	}

	inst, err := fluxo.Run(ctx, eng, flow.Name(), "Gopher")
	if err != nil {
		log.Fatal(err)
	}

	fmt.Printf("workflow %q finished with status %s and output %v\n",
		inst.ID, inst.Status, inst.Output)
}

func sayHello(ctx context.Context, input any) (any, error) {
	name, ok := input.(string)
	if !ok {
		return nil, fmt.Errorf("sayHello: expected string input, got %T", input)
	}
	msg := fmt.Sprintf("hello, %s", name)
	log.Printf("[sayHello] %s", msg)
	return msg, nil
}

func decorateMessage(ctx context.Context, input any) (any, error) {
	msg, ok := input.(string)
	if !ok {
		return nil, fmt.Errorf("decorateMessage: expected string input, got %T", input)
	}
	out := fmt.Sprintf("*** %s ***", msg)
	log.Printf("[decorateMessage] %s", out)
	return out, nil
}
Example (LocalRunner)

Example_localRunner demonstrates using LocalRunner to execute workflows with an in-process engine, queue, and worker. 

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/petrijr/fluxo"
)

func main() {
	ctx := context.Background()

	runner := fluxo.NewLocalRunner()

	flow := fluxo.New("Greeting").
		Step("sayHello", sayHello).
		Step("decorateMessage", decorateMessage)

	if err := flow.Register(runner.Engine); err != nil {
		log.Fatal(err)
	}

	// Start one worker goroutine.
	if err := runner.StartWorkers(ctx, 1); err != nil {
		log.Fatal(err)
	}
	defer runner.Stop()

	// Enqueue an asynchronous workflow start.
	if err := runner.StartWorkflowAsync(ctx, flow.Name(), "Gopher"); err != nil {
		log.Fatal(err)
	}

	// In a real application you'd wait on instance completion or poll;
	// for example purposes, just give the worker a moment to run.
	time.Sleep(500 * time.Millisecond)
}

func sayHello(ctx context.Context, input any) (any, error) {
	name, ok := input.(string)
	if !ok {
		return nil, fmt.Errorf("sayHello: expected string input, got %T", input)
	}
	msg := fmt.Sprintf("hello, %s", name)
	log.Printf("[sayHello] %s", msg)
	return msg, nil
}

func decorateMessage(ctx context.Context, input any) (any, error) {
	msg, ok := input.(string)
	if !ok {
		return nil, fmt.Errorf("decorateMessage: expected string input, got %T", input)
	}
	out := fmt.Sprintf("*** %s ***", msg)
	log.Printf("[decorateMessage] %s", out)
	return out, nil
}

Index

Examples

Constants

View Source 
const (
	StatusPending   = api.StatusPending
	StatusRunning   = api.StatusRunning
	StatusWaiting   = api.StatusWaiting
	StatusFailed    = api.StatusFailed
	StatusCompleted = api.StatusCompleted
)

Re-export status values for convenience.

Variables

View Source 
var (
	NewLoggingObserver   = api.NewLoggingObserver
	NewCompositeObserver = api.NewCompositeObserver
)

Re-export common observer helpers.

Functions

func RecoverStuckInstances 

func RecoverStuckInstances(ctx context.Context, eng Engine) (int, error)

RecoverStuckInstances delegates to eng.RecoverStuckInstances.

It is typically called on process startup before starting any workers: 

count, err := fluxo.RecoverStuckInstances(ctx, engine)

Types

type BasicMetrics 

type BasicMetrics = api.BasicMetrics

Re-export key types so users don't need to dig into pkg/api.

type BasicMetricsSnapshot 

type BasicMetricsSnapshot = api.BasicMetricsSnapshot

Re-export key types so users don't need to dig into pkg/api.

type ChildWorkflowSpec 

type ChildWorkflowSpec = api.ChildWorkflowSpec

Re-export key types so users don't need to dig into pkg/api.

type CompositeObserver 

type CompositeObserver = api.CompositeObserver

Re-export key types so users don't need to dig into pkg/api.

type ConditionFunc 

type ConditionFunc = api.ConditionFunc

Re-export key types so users don't need to dig into pkg/api.

type Config 

type Config = worker.Config

Re-export key types so users don't need to dig into pkg/api.

type Engine 

type Engine = api.Engine

Re-export key types so users don't need to dig into pkg/api.

func NewInMemoryEngine 

func NewInMemoryEngine() Engine

NewInMemoryEngine returns an Engine backed entirely by in-memory stores.

func NewInMemoryEngineWithObserver 

func NewInMemoryEngineWithObserver(obs Observer) Engine

NewInMemoryEngineWithObserver returns an in-memory Engine with the given Observer.

func NewSQLiteEngine 

func NewSQLiteEngine(db *sql.DB) (Engine, error)

NewSQLiteEngine returns an Engine that persists workflow instances in a SQLite database. Workflow definitions are kept in-memory.

func NewSQLiteEngineWithObserver 

func NewSQLiteEngineWithObserver(db *sql.DB, obs Observer) (Engine, error)

NewSQLiteEngineWithObserver returns a SQLite-backed Engine with the given Observer.

type FlowBuilder 

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

FlowBuilder provides a fluent API for defining workflows: 

flow := fluxo.New("OnboardUser").
    Step("createAccount", createAccount).
    Step("sendWelcomeEmail", sendWelcomeEmail).
    Step("waitActivation", fluxo.WaitForSignalStep("activated"))

if err := flow.Register(engine); err != nil {
    log.Fatal(err)
}

inst, err := fluxo.Run(ctx, engine, flow.Name(), input)

func New 

func New(name string) *FlowBuilder

New creates a new workflow builder with the given name.

func (*FlowBuilder) Definition 

func (b *FlowBuilder) Definition() WorkflowDefinition

Definition returns the underlying WorkflowDefinition. Typically used when interacting with lower-level APIs.

func (*FlowBuilder) If 

func (b *FlowBuilder) If(name string, cond ConditionFunc, thenStep, elseStep StepFunc) *FlowBuilder

If adds a conditional branching step.

func (*FlowBuilder) Loop 

func (b *FlowBuilder) Loop(name string, times int, body StepFunc) *FlowBuilder

Loop adds a step that executes body a fixed number of times. The loop is executed as a nested step; retries/backoff (if any) apply to the entire loop execution.

func (*FlowBuilder) MustRegister 

func (b *FlowBuilder) MustRegister(eng Engine)

MustRegister is like Register but panics on error. Useful for initialization in main().

func (*FlowBuilder) Name 

func (b *FlowBuilder) Name() string

Name returns the workflow name.

func (*FlowBuilder) Parallel 

func (b *FlowBuilder) Parallel(name string, steps ...StepFunc) *FlowBuilder

Parallel is a convenience for adding a step that runs sub-steps in parallel.

func (*FlowBuilder) Register 

func (b *FlowBuilder) Register(eng Engine) error

Register registers the built workflow with the given engine.

func (*FlowBuilder) Step 

func (b *FlowBuilder) Step(name string, fn StepFunc) *FlowBuilder

Step appends a basic step to the workflow.

func (*FlowBuilder) StepWithRetry 

func (b *FlowBuilder) StepWithRetry(name string, fn StepFunc, retry RetryPolicy) *FlowBuilder

StepWithRetry appends a step that uses the given retry policy.

func (*FlowBuilder) StepWithRetryBuilder 

func (b *FlowBuilder) StepWithRetryBuilder(name string, fn StepFunc, rb RetryBuilder) *FlowBuilder

StepWithRetryBuilder is a convenience wrapper around StepWithRetry that accepts a RetryBuilder.

func (*FlowBuilder) Switch 

func (b *FlowBuilder) Switch(
	name string,
	selector SelectorFunc,
	branches map[string]StepFunc,
	defaultStep StepFunc,
) *FlowBuilder

Switch adds a multi-branch step based on a selector and branch map.

func (*FlowBuilder) WaitForAnySignal 

func (b *FlowBuilder) WaitForAnySignal(stepName string, names ...string) *FlowBuilder

WaitForAnySignal adds a step that waits for any of the given signal names.

func (*FlowBuilder) WaitForSignal 

func (b *FlowBuilder) WaitForSignal(stepName, signalName string) *FlowBuilder

WaitForSignal adds a step that waits for a named signal.

func (*FlowBuilder) While 

func (b *FlowBuilder) While(name string, cond ConditionFunc, body StepFunc) *FlowBuilder

While adds a looping step that executes body while cond(input) is true. The loop is executed as a nested step; retries/backoff (if any) apply to the entire loop execution.

type InstanceListOptions 

type InstanceListOptions = api.InstanceListOptions

Re-export key types so users don't need to dig into pkg/api.

type LocalRunner 

type LocalRunner struct {
	// Engine is the in-memory workflow engine used by this runner.
	Engine Engine

	// Queue is the in-memory task queue used by the Worker.
	Queue taskqueue.Queue

	// Worker processes tasks from Queue using Engine.
	Worker *worker.Worker
	// contains filtered or unexported fields
}

LocalRunner bundles an in-memory Engine, an in-memory task queue, and a Worker to provide a simple "local runner" for development and debugging.

Typical usage: 

runner := fluxo.NewLocalRunner()
flow := fluxo.New("my-flow").Step(...)
flow.MustRegister(runner.Engine)

// Synchronous run (no queue/worker involved):
inst, err := fluxo.Run(ctx, runner.Engine, flow.Name(), input)

// Asynchronous run:
_ = runner.StartWorkers(ctx, 2)
_ = runner.StartWorkflowAsync(ctx, flow.Name(), input)
...
runner.Stop()

func NewLocalRunner 

func NewLocalRunner() *LocalRunner

NewLocalRunner constructs a LocalRunner backed by an in-memory engine, in-memory queue, and a Worker with default config.

This is intended for local development, tests, and simple single-process deployments.

func (*LocalRunner) SignalAsync 

func (r *LocalRunner) SignalAsync(ctx context.Context, instanceID, name string, payload any) error

SignalAsync enqueues a task to deliver a signal to a workflow instance. The instance will process the signal when a worker picks up the task.

func (*LocalRunner) StartWorkers 

func (r *LocalRunner) StartWorkers(ctx context.Context, concurrency int) error

StartWorkers starts 'concurrency' worker goroutines that continuously call Worker.ProcessOne(ctx) until the context is cancelled via Stop.

If StartWorkers is called more than once without Stop, it returns an error.

func (*LocalRunner) StartWorkflowAsync 

func (r *LocalRunner) StartWorkflowAsync(ctx context.Context, workflowName string, input any) error

StartWorkflowAsync enqueues a task to start the given workflow asynchronously. The workflow must already be registered on LocalRunner.Engine.

func (*LocalRunner) Stop 

func (r *LocalRunner) Stop()

Stop cancels all worker goroutines started by StartWorkers and waits for them to exit.

type LoggingObserver 

type LoggingObserver = api.LoggingObserver

Re-export key types so users don't need to dig into pkg/api.

type NoopObserver 

type NoopObserver = api.NoopObserver

Re-export key types so users don't need to dig into pkg/api.

type Observer 

type Observer = api.Observer

Re-export key types so users don't need to dig into pkg/api.

type ParallelResult 

type ParallelResult = api.ParallelResult

Re-export key types so users don't need to dig into pkg/api.

type Queue 

type Queue = taskqueue.Queue

Re-export key types so users don't need to dig into pkg/api.

func NewInMemoryQueue 

func NewInMemoryQueue(size int) Queue

func NewSQLiteQueue 

func NewSQLiteQueue(db *sql.DB) (Queue, error)

type RetryBuilder 

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

RetryBuilder provides a fluent way to construct RetryPolicy values for use with FlowBuilder.StepWithRetry.

func Retry 

func Retry(maxAttempts int) RetryBuilder

Retry creates a RetryBuilder with the given maxAttempts.

maxAttempts <= 0 is treated as 1 (no retries).

func (RetryBuilder) Immediate 

func (r RetryBuilder) Immediate() RetryBuilder

Immediate disables any sleep between retries. Retries will still respect MaxAttempts.

func (RetryBuilder) Policy 

func (r RetryBuilder) Policy() RetryPolicy

Policy returns the underlying RetryPolicy to be passed to FlowBuilder.StepWithRetry.

func (RetryBuilder) WithConstantBackoff 

func (r RetryBuilder) WithConstantBackoff(delay time.Duration) RetryBuilder

WithConstantBackoff configures a constant backoff between retries.

This is equivalent to an exponential backoff with multiplier 1.0 and no max cap.

func (RetryBuilder) WithExponentialBackoff 

func (r RetryBuilder) WithExponentialBackoff(initial time.Duration, multiplier float64, max time.Duration) RetryBuilder

WithExponentialBackoff configures exponential backoff:

  • initial is the delay before the first retry.
  • multiplier > 1 grows the delay each attempt (default 2.0 if <= 0).
  • max caps the delay; if <= 0, there is no cap.

Example: 

Retry(3).WithExponentialBackoff(100*time.Millisecond, 2.0, 2*time.Second)

type RetryPolicy 

type RetryPolicy = api.RetryPolicy

Re-export key types so users don't need to dig into pkg/api.

type SelectorFunc 

type SelectorFunc = api.SelectorFunc

Re-export key types so users don't need to dig into pkg/api.

type Status 

type Status = api.Status

Re-export key types so users don't need to dig into pkg/api.

type StepDefinition 

type StepDefinition = api.StepDefinition

Re-export key types so users don't need to dig into pkg/api.

type StepFunc 

type StepFunc = api.StepFunc

Re-export key types so users don't need to dig into pkg/api.

func IfStep 

func IfStep(cond ConditionFunc, thenStep, elseStep StepFunc) StepFunc

IfStep creates a conditional step composed of then/else branches.

func LoopStep 

func LoopStep(times int, body StepFunc) StepFunc

LoopStep returns a step that executes body a fixed number of times. The entire loop is treated as a single engine step.

func ParallelMapStep 

func ParallelMapStep(mapper StepFunc) StepFunc

ParallelMapStep runs a mapping step over a slice input in parallel.

func ParallelStep 

func ParallelStep(steps ...StepFunc) StepFunc

ParallelStep runs all provided step funcs in parallel and returns a []any of their outputs.

func SleepStep 

func SleepStep(d time.Duration) StepFunc

SleepStep returns a step that sleeps for the given duration and passes the input through.

func SleepUntilStep 

func SleepUntilStep(t time.Time) StepFunc

SleepUntilStep sleeps until a given timestamp or returns ctx.Err.

func StartChildrenStep 

func StartChildrenStep(specsFn func(input any) ([]api.ChildWorkflowSpec, error)) StepFunc

StartChildrenStep starts child workflows and returns their IDs.

func SwitchStep 

func SwitchStep(selector SelectorFunc, branches map[string]StepFunc, defaultStep StepFunc) StepFunc

SwitchStep dispatches to a branch based on a selector.

func TypedLoop 

func TypedLoop[I any](times int, body func(context.Context, I) (I, error)) StepFunc

TypedLoop returns a step that executes a strongly-typed body a fixed number of times.

func TypedStep 

func TypedStep[I, O any](fn func(context.Context, I) (O, error)) StepFunc

TypedStep wraps a strongly-typed function into a StepFunc. Example: 

fluxo.TypedStep(func(ctx context.Context, s MyState) (MyState, error) { ... })

func TypedWhile 

func TypedWhile[I any](cond func(I) bool, body func(context.Context, I) (I, error)) StepFunc

TypedWhile returns a step that repeatedly executes a strongly-typed body while cond(input) is true.

func WaitForAnyChildStep 

func WaitForAnyChildStep(getIDs func(input any) []string, pollInterval time.Duration) StepFunc

WaitForAnyChildStep waits until any of the children completes.

func WaitForAnySignalStep 

func WaitForAnySignalStep(names ...string) StepFunc

WaitForAnySignalStep waits for one of the allowed signal names.

func WaitForChildrenStep 

func WaitForChildrenStep(getIDs func(input any) []string, pollInterval time.Duration) StepFunc

WaitForChildrenStep waits for all given child workflow IDs to complete.

func WaitForSignalStep 

func WaitForSignalStep(name string) StepFunc

WaitForSignalStep waits for a single named signal, returning its payload.

func While 

func While(cond ConditionFunc, body StepFunc) StepFunc

While returns a step that repeatedly executes body while cond(input) is true. The entire loop is treated as a single engine step.

type TimeoutPayload 

type TimeoutPayload = api.TimeoutPayload

Re-export key types so users don't need to dig into pkg/api.

type Worker 

type Worker = worker.Worker

Re-export key types so users don't need to dig into pkg/api.

func NewWorker 

func NewWorker(engine api.Engine, queue Queue) *Worker

func NewWorkerWithConfig 

func NewWorkerWithConfig(engine api.Engine, queue Queue, cfg Config) *Worker

type WorkerBundle 

type WorkerBundle struct {
	Engine Engine
	Worker *workerpkg.Worker
	// contains filtered or unexported fields
}

WorkerBundle wires together an Engine, a durable task queue, and a Worker that consumes tasks from that queue.

For now, we only provide a SQLite-backed bundle.

func NewSQLiteBundle 

func NewSQLiteBundle(db *sql.DB, cfg workerpkg.Config) (*WorkerBundle, error)

NewSQLiteBundle constructs a durable Engine + Queue + Worker combo sharing the same SQLite database. Workflow instances and queued tasks are persisted in the provided *sql.DB.

Typical usage: 

db, _ := sql.Open("sqlite", "file:fluxo.db?_journal=WAL")
bundle, err := fluxo.NewSQLiteBundle(db, worker.Config{MaxAttempts: 3})
// register workflows on bundle.Engine
// enqueue work via bundle.Worker

type WorkflowDefinition 

type WorkflowDefinition = api.WorkflowDefinition

Re-export key types so users don't need to dig into pkg/api.

type WorkflowInstance 

type WorkflowInstance = api.WorkflowInstance

Re-export key types so users don't need to dig into pkg/api.

func GetInstance 

func GetInstance(ctx context.Context, eng Engine, id string) (*WorkflowInstance, error)

GetInstance fetches an instance by ID.

func ListInstances 

func ListInstances(ctx context.Context, eng Engine, opts InstanceListOptions) ([]*WorkflowInstance, error)

ListInstances lists workflow instances according to the given options.

func Resume 

func Resume(ctx context.Context, eng Engine, id string) (*WorkflowInstance, error)

Resume resumes a previously failed instance.

func Run 

func Run(ctx context.Context, eng Engine, name string, input any) (*WorkflowInstance, error)

Run runs a registered workflow synchronously.

func Signal 

func Signal(ctx context.Context, eng Engine, id string, name string, payload any) (*WorkflowInstance, error)

Signal delivers a signal to a waiting instance.

Source Files

View all Source files

Directories

Path Synopsis
examples
approval command
approval_http command
builder command
examples/builder/main.go
 examples/builder/main.go
localrunner command
loop command
observer command
examples/observer/main.go
 examples/observer/main.go
parallel command
retry command
signal_timeout command
sqlite_bundle command
typed command
examples/typed/main.go
 examples/typed/main.go
internal
engine
persistence
taskqueue
pkg
api
Package api contains the core building blocks used by the fluxo workflow engine.
 Package api contains the core building blocks used by the fluxo workflow engine.
worker
Package worker provides the background worker implementation used to drive fluxo workflows forward.
 Package worker provides the background worker implementation used to drive fluxo workflows forward.

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.