No-code workflow executor. it executes DAGs defined in a simple YAML format

Dagu

A taskflow-like General-purpose Task-parallel Programming Framework with integrated visualizer and profiler

# go-taskflow

[![codecov](https://codecov.io/github/noneback/go-taskflow/graph/badge.svg?token=CITXYA10C6)](https://codecov.io/github/noneback/go-taskflow)
[![Go Reference](https://pkg.go.dev/badge/github.com/noneback/go-taskflow.svg)](https://pkg.go.dev/github.com/noneback/go-taskflow)
[![Go Report Card](https://goreportcard.com/badge/github.com/noneback/go-taskflow)](https://goreportcard.com/report/github.com/noneback/go-taskflow)
[![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go)
[![DeepWiki][deepwiki-image]][deepwiki-url]

[deepwiki-url]: https://deepwiki.com/noneback/go-taskflow
[deepwiki-image]: https://img.shields.io/badge/Chat%20with-DeepWiki%20🤖-20B2AA

![go-taskflow](https://socialify.git.ci/noneback/go-taskflow/image?description=1&language=1&name=1&pattern=Solid&theme=Auto)

go-taskflow is a general-purpose task-parallel programming framework for Go, inspired by [taskflow-cpp](https://github.com/taskflow/taskflow). It leverages Go's native capabilities and simplicity, making it ideal for managing complex dependencies in concurrent tasks.

## Features

- **High Extensibility**: Easily extend the framework to adapt to various specific use cases.
- **Native Go Concurrency Model**: Leverages Go's goroutines for efficient concurrent task execution.
- **User-Friendly Programming Interface**: Simplifies complex task dependency management in Go.
- **Static, Subflow, Conditional, and Cyclic Tasking**: Define static tasks, condition nodes, nested subflows, and cyclic flows to enhance modularity and programmability.

    | Static | Subflow | Condition | Cyclic |
    |:-----------|:------------:|------------:|------------:|
    | ![](image/simple.svg)     |   ![](image/subflow.svg)   |      ![](image/condition.svg) |      ![](image/loop.svg) |

- **Priority Task Scheduling**: Assign task priorities to ensure higher-priority tasks are executed first.
- **Built-in Visualization, Profiling and Tracing**: Generate visual representations of tasks, profile task execution with flamegraph, and capture Chrome Trace events for timeline analysis.

## Use Cases

- **Data Pipelines**: Orchestrate data processing stages with complex dependencies.
- **AI Agent Workflow Automation**: Define and execute AI agent workflows with clear sequences and dependency structures.
- **Parallel Graph Tasking**: Execute graph-based tasks concurrently to maximize CPU utilization.

## Installation

Import the latest version of go-taskflow using:

```bash
go get -u github.com/noneback/go-taskflow
```

## Documentation

[DeepWiki Page](https://deepwiki.com/noneback/go-taskflow)

## Example

Below is an example of using go-taskflow to implement a parallel merge sort:

```go
package main

import (
    "fmt"
    "log"
    "math/rand"
    "os"
    "slices"
    "strconv"
    "sync"

    gtf "github.com/noneback/go-taskflow"
)

// mergeInto merges a sorted source array into a sorted destination array.
func mergeInto(dest, src []int) []int {
    size := len(dest) + len(src)
    tmp := make([]int, 0, size)
    i, j := 0, 0
    for i < len(dest) && j < len(src) {
        if dest[i] < src[j] {
            tmp = append(tmp, dest[i])
            i++
        } else {
            tmp = append(tmp, src[j])
            j++
        }
    }

    if i < len(dest) {
        tmp = append(tmp, dest[i:]...)
    } else {
        tmp = append(tmp, src[j:]...)
    }

    return tmp
}

func main() {
    size := 100
    randomArr := make([][]int, 10)
    sortedArr := make([]int, 0, 10*size)
    mutex := &sync.Mutex{}

    for i := 0; i < 10; i++ {
        for j := 0; j < size; j++ {
            randomArr[i] = append(randomArr[i], rand.Int())
        }
    }

    sortTasks := make([]*gtf.Task, 10)
    tf := gtf.NewTaskFlow("merge sort")
    done := tf.NewTask("Done", func() {
        if !slices.IsSorted(sortedArr) {
            log.Fatal("Sorting failed")
        }
        fmt.Println("Sorted successfully")
        fmt.Println(sortedArr[:1000])
    })

    for i := 0; i < 10; i++ {
        sortTasks[i] = tf.NewTask("sort_"+strconv.Itoa(i), func() {
            arr := randomArr[i]
            slices.Sort(arr)
            mutex.Lock()
            defer mutex.Unlock()
            sortedArr = mergeInto(sortedArr, arr)
        })
    }
    done.Succeed(sortTasks...)

    executor := gtf.NewExecutor(1000, gtf.WithProfiler())

    executor.Run(tf).Wait()

    if err := tf.Dump(os.Stdout); err != nil {
        log.Fatal("Error dumping taskflow:", err)
    }

    if err := executor.Profile(os.Stdout); err != nil {
        log.Fatal("Error profiling taskflow:", err)
    }
}
```

For more examples, visit the [examples directory](https://github.com/noneback/go-taskflow/tree/main/examples).

## Benchmark

The following benchmark provides a rough estimate of performance. Note that most realistic workloads are I/O-bound, and their performance cannot be accurately reflected by these results. For CPU-intensive tasks, consider using [taskflow-cpp](https://github.com/taskflow/taskflow).

```plaintext
$ go test -bench=. -benchmem
goos: linux
goarch: amd64
pkg: github.com/noneback/go-taskflow/benchmark
cpu: Intel(R) Xeon(R) Platinum 8269CY CPU @ 2.50GHz
BenchmarkC32-4    	   23282	     51891 ns/op	    7295 B/op	     227 allocs/op
BenchmarkS32-4    	    7047	    160199 ns/op	    6907 B/op	     255 allocs/op
BenchmarkC6-4     	   66397	     18289 ns/op	    1296 B/op	      47 allocs/op
BenchmarkC8x8-4   	    7946	    143474 ns/op	   16914 B/op	     504 allocs/op
PASS
ok  	github.com/noneback/go-taskflow/benchmark	5.606s
```

## Understanding Conditional Tasks

Conditional nodes in go-taskflow behave similarly to those in [taskflow-cpp](https://github.com/taskflow/taskflow). They participate in both conditional control and looping. To avoid common pitfalls, refer to the [Conditional Tasking documentation](https://taskflow.github.io/taskflow/ConditionalTasking.html).

## Executor Options

`NewExecutor` accepts functional options to configure behavior:

```go
executor := gtf.NewExecutor(1000,
    gtf.WithProfiler(), // enable flamegraph profiling
    gtf.WithTracer(),   // enable Chrome Trace recording
)
```

| Option | Description |
|:---|:---|
| `WithProfiler()` | Enable flamegraph profiling. Required before calling `executor.Profile()`. |
| `WithTracer()` | Enable Chrome Trace recording. Required before calling `executor.Trace()`. |

## Error Handling in go-taskflow

In Go, `errors` are values, and it is the user's responsibility to handle them appropriately. Only unrecovered `panic` events are managed by the framework. If a `panic` occurs, the entire parent graph is canceled, leaving the remaining tasks incomplete. This behavior may evolve in the future. If you have suggestions, feel free to share them.

To prevent interruptions caused by `panic`, you can handle them manually when registering tasks:

```go
tf.NewTask("not interrupt", func() {
    defer func() {
        if r := recover(); r != nil {
            // Handle the panic.
        }
    }()
    // User-defined logic.
})
```

## Visualizing Taskflows

To generate a visual representation of a taskflow, use the `Dump` method:

```go
if err := tf.Dump(os.Stdout); err != nil {
    log.Fatal(err)
}
```

The `Dump` method generates raw strings in DOT format. Use the `dot` tool to create a graph SVG. 

![dot](image/desc.svg)

## Profiling Taskflows

To profile a taskflow, first enable the profiler with `WithProfiler()`, then call `Profile`:

```go
executor := gtf.NewExecutor(1000, gtf.WithProfiler())
executor.Run(tf).Wait()

if err := executor.Profile(os.Stdout); err != nil {
    log.Fatal(err)
}
```

The `Profile` method generates raw strings in flamegraph format. Use the `flamegraph` tool to create a flamegraph SVG.

![flg](image/fl.svg)

## Tracing Taskflows

To trace a taskflow, first enable the tracer with `WithTracer()`, then call `Trace`:

```go
executor := gtf.NewExecutor(1000, gtf.WithTracer())
executor.Run(tf).Wait()

if err := executor.Trace(os.Stdout); err != nil {
    log.Fatal(err)
}
```

The `Trace` method outputs JSON in [Chrome Trace Event format](https://docs.google.com/document/d/1CvAClvFfyA5R-PhYUmn5OOQtYMH4h6I0nSsKchNAySU). Open it in `chrome://tracing` or [Perfetto UI](https://ui.perfetto.dev/) for visualization.

## Stargazer

[![Star History Chart](https://api.star-history.com/svg?repos=noneback/go-taskflow&type=Date)](https://star-history.com/#noneback/go-taskflow&Date)



go-taskflow

A tech stack agnostic Event Driven Workflow framework

<div align="center">
    <img src="./logo/logo.png" style="width: 220px; margin: 30px" alt="Workflow Logo">
    <div  align="center" style="max-width: 750px">
        <a style="padding: 0 5px" href="https://goreportcard.com/report/github.com/luno/workflow"><img src="https://goreportcard.com/badge/github.com/luno/workflow"/></a>
        <a style="padding: 0 5px" href="https://sonarcloud.io/summary/new_code?id=luno_workflow"><img src="https://sonarcloud.io/api/project_badges/measure?project=luno_workflow&metric=coverage"/></a>
        <a style="padding: 0 5px" href="https://sonarcloud.io/summary/new_code?id=luno_workflow"><img src="https://sonarcloud.io/api/project_badges/measure?project=luno_workflow&metric=alert_status"/></a>
        <a style="padding: 0 5px" href="https://sonarcloud.io/summary/new_code?id=luno_workflow"><img src="https://sonarcloud.io/api/project_badges/measure?project=luno_workflow&metric=bugs"/></a>
        <a style="padding: 0 5px" href="https://sonarcloud.io/summary/new_code?id=luno_workflow"><img src="https://sonarcloud.io/api/project_badges/measure?project=luno_workflow&metric=reliability_rating"/></a><a style="text-decoration:none; padding: 0 5px" href="https://sonarcloud.io/summary/new_code?id=luno_workflow" ><img src="https://sonarcloud.io/api/project_badges/measure?project=luno_workflow&metric=security_rating"/></a>
        <a style="padding: 0 5px" href="https://sonarcloud.io/summary/new_code?id=luno_workflow"><img src="https://sonarcloud.io/api/project_badges/measure?project=luno_workflow&metric=sqale_rating"/></a>
        <a style="padding: 0 5px" href="https://sonarcloud.io/summary/new_code?id=luno_workflow"><img src="https://sonarcloud.io/api/project_badges/measure?project=luno_workflow&metric=vulnerabilities"/></a>
        <a style="padding: 0 5px" href="https://sonarcloud.io/summary/new_code?id=luno_workflow"><img src="https://sonarcloud.io/api/project_badges/measure?project=luno_workflow&metric=duplicated_lines_density"/></a>
        <a style="padding: 0 5px" href="https://pkg.go.dev/github.com/luno/workflow"><img src="https://pkg.go.dev/badge/github.com/luno/workflow.svg" alt="Go Reference"></a>
        <a style="padding: 0 5px" href="https://github.com/avelino/awesome-go"><img src="https://awesome.re/mentioned-badge-flat.svg" alt="Mentioned in Awesome Go"></a>
    </div>
</div>

# Workflow

**The type-safe, event-driven workflow orchestration library that scales with your business.**

Build robust, distributed workflows in Go with compile-time safety, automatic retries, and horizontal scaling out of the box.

```go
// Define your business logic as type-safe state machines
b := workflow.NewBuilder[Order, OrderStatus]("order-processing")
b.AddStep(OrderCreated, ProcessPayment, PaymentProcessed)
b.AddStep(PaymentProcessed, FulfillOrder, OrderCompleted)

wf := b.Build(kafkaStreamer, sqlStore, roleScheduler)
```

## Why Choose Workflow?

### 🎯 **Type-Safe by Design**
Unlike other orchestrators, Workflow leverages Go generics for **compile-time guarantees**. Catch errors before deployment, not in production.

```go
// Your IDE knows exactly what data flows where
func processPayment(ctx context.Context, r *workflow.Run[Order, OrderStatus]) (OrderStatus, error) {
    // r.Object is typed as *Order, OrderStatus is your enum
    // Compiler catches mismatches before they cause runtime errors
}
```

### ⚡ **Event-Driven Architecture**
Built for modern distributed systems. Steps communicate through durable events, enabling:
- **Loose coupling** between workflow components
- **Automatic retries** with exponential backoff
- **Horizontal scaling** across multiple instances
- **Fault tolerance** that survives network partitions

### 🔧 **Infrastructure Agnostic**
Your choice of database, message queue, and coordination service. Start simple, scale when needed:

```go
// Development: Everything in-memory
wf := b.Build(memstreamer.New(), memrecordstore.New(), memrolescheduler.New())

// Production: Battle-tested infrastructure
wf := b.Build(kafkastreamer.New(), sqlstore.New(), rinkrolescheduler.New())
```

### 📊 **Built-in Observability**
Production-ready monitoring without the setup overhead:
- **Prometheus metrics** for throughput, latency, and error rates
- **Web UI** for real-time workflow visualization
- **Structured logging** with correlation IDs
- **Distributed tracing** support

## Perfect For

- **Order Processing**: Payment, inventory, fulfillment pipelines
- **User Onboarding**: Multi-step verification and activation flows
- **Financial Operations**: Transaction processing with compliance checks
- **Data Processing**: ETL pipelines with validation and cleanup
- **Approval Workflows**: Multi-stakeholder review processes

## vs. The Alternatives

| Feature | **Workflow** | Temporal | Zeebe/Camunda |
|---------|-------------|----------|---------------|
| **Type Safety** | ✅ Compile-time (Go generics) | ❌ Runtime validation | ❌ Runtime (BPMN) |
| **Architecture** | ✅ Event-driven state machines | ⚠️ RPC-based activities | ⚠️ Token-based execution |
| **Infrastructure** | ✅ Your choice (adapters) | ❌ Requires Temporal cluster | ❌ Requires external engine |
| **Deployment** | ✅ Library in your app | ❌ Separate server/workers | ❌ Separate engine |
| **Learning Curve** | ✅ Native Go patterns | ⚠️ New concepts & SDKs | ❌ BPMN modeling |
| **Language** | ✅ Go-native | ⚠️ Multi-language via gRPC | ⚠️ Multi-language |

## Quick Start

```bash
go get github.com/luno/workflow
```

```go
package main

import (
    "context"
    "fmt"
    "github.com/luno/workflow"
    "github.com/luno/workflow/adapters/memstreamer"
    "github.com/luno/workflow/adapters/memrecordstore"
    "github.com/luno/workflow/adapters/memrolescheduler"
)

type TaskStatus int
const (
    TaskStatusUnknown   TaskStatus = 0
    TaskStatusCreated   TaskStatus = 1
    TaskStatusProcessed TaskStatus = 2
    TaskStatusCompleted TaskStatus = 3
)

type Task struct {
    ID   string
    Name string
}

func main() {
    b := workflow.NewBuilder[Task, TaskStatus]("task-processor")

    b.AddStep(TaskStatusCreated, func(ctx context.Context, r *workflow.Run[Task, TaskStatus]) (TaskStatus, error) {
        fmt.Printf("Processing: %s\n", r.Object.Name)
        return TaskStatusProcessed, nil
    }, TaskStatusProcessed)

    b.AddStep(TaskStatusProcessed, func(ctx context.Context, r *workflow.Run[Task, TaskStatus]) (TaskStatus, error) {
        fmt.Printf("Completed: %s\n", r.Object.Name)
        return TaskStatusCompleted, nil
    }, TaskStatusCompleted)

    wf := b.Build(memstreamer.New(), memrecordstore.New(), memrolescheduler.New())

    ctx := context.Background()
    wf.Run(ctx)
    defer wf.Stop()

    // Trigger a workflow
    runID, _ := wf.Trigger(ctx, "task-1", workflow.WithInitialValue(&Task{
        ID: "task-1",
        Name: "Process Invoice",
    }))

    // Wait for completion
    wf.Await(ctx, "task-1", runID, TaskStatusCompleted)
    fmt.Println("✅ Workflow completed!")
}
```

## Enterprise Ready

**Workflow** provides enterprise-grade features:

- ✅ **Exactly-once processing** guarantees via transactional outbox pattern
- ✅ **Built-in error handling** with pause and retry mechanisms
- ✅ **Comprehensive observability** via Prometheus metrics and Web UI
- ✅ **Horizontal scaling** through role-based scheduling
- ✅ **Infrastructure flexibility** via pluggable adapters
- ✅ **Production deployment** patterns for various scales

## Documentation

| Topic | Description |
|-------|-------------|
| **[Getting Started](docs/getting-started.md)** | Install and build your first workflow |
| **[Core Concepts](docs/concepts.md)** | Understand Runs, Events, and State Machines |
| **[Architecture](docs/architecture.md)** | Deep dive into system design and components |
| **[Steps](docs/steps.md)** | Build workflow logic with step functions |
| **[Adapters](docs/adapters.md)** | Infrastructure integration guide |
| **[Database Setup](docs/database-setup.md)** | Complete MariaDB/MySQL & PostgreSQL setup guide |
| **[Callbacks](docs/callbacks.md)** | Handle external events and webhooks |
| **[Timeouts](docs/timeouts.md)** | Add time-based operations |
| **[Connectors](docs/connectors.md)** | Integrate with external event streams |
| **[Hooks](docs/hooks.md)** | React to workflow lifecycle changes |
| **[Configuration](docs/configuration.md)** | Tune performance and behaviour |
| **[Monitoring](docs/monitoring.md)** | Observability and debugging |

### Examples & Tutorials

| Example | Description |
|---------|-------------|
| **[Getting Started](docs/getting-started.md)** | Your first workflow in 5 minutes |
| **[SQL Example](_examples/sqlexample)** | Complete MariaDB/MySQL integration with Docker |
| **[Order Processing](_examples/orderprocessor)** | Complex e-commerce workflow with payments & fulfilment |
| **[Callbacks](_examples/callback)** | Handle external webhooks and events |
| **[Timeouts](_examples/timeout)** | Schedule and timeout operations |
| **[Connectors](_examples/connector)** | Integrate with external event streams |


## Community & Support

- 📚 **[Documentation](docs/)** - Comprehensive guides and examples
- 🐛 **[Issues](https://github.com/luno/workflow/issues)** - Bug reports and feature requests
- 💬 **[Discussions](https://github.com/luno/workflow/discussions)** - Community Q&A

## Installation

```bash
go get github.com/luno/workflow

# Production adapters (install as needed)
go get github.com/luno/workflow/adapters/kafkastreamer
go get github.com/luno/workflow/adapters/sqlstore
go get github.com/luno/workflow/adapters/rinkrolescheduler
go get github.com/luno/workflow/adapters/webui
```

## License

[MIT License](LICENSE)

---

**Ready to build reliable workflows?** [Get started in 5 minutes →](docs/getting-started.md)

workflow

A framework for authoring workflows and activities running on top of the Cadence orchestration engine made by Uber

# Go framework for Cadence
[![CI Checks](https://github.com/cadence-workflow/cadence-go-client/actions/workflows/ci-checks.yml/badge.svg)](https://github.com/cadence-workflow/cadence-go-client/actions/workflows/ci-checks.yml)
[![Coverage](https://codecov.io/gh/uber-go/cadence-client/graph/badge.svg?token=iEpqo5HbDe)](https://codecov.io/gh/uber-go/cadence-client)
[![Slack Status](https://img.shields.io/badge/slack-join_chat-white.svg?logo=slack&style=social)](https://communityinviter.com/apps/cloud-native/cncf)
[![GoDoc](https://godoc.org/go.uber.org/cadence?status.svg)](https://godoc.org/go.uber.org/cadence)

[Cadence](https://github.com/uber/cadence) is a distributed, scalable, durable, and highly available orchestration engine we developed at Uber Engineering to execute asynchronous long-running business logic in a scalable and resilient way.

`cadence-client` is the framework for authoring workflows and activities.

## How to use

Make sure you clone this repo into the correct location.

```bash
git clone git@github.com:uber-go/cadence-client.git $GOPATH/src/go.uber.org/cadence
```

or

```bash
go get go.uber.org/cadence
```

See [samples](https://github.com/uber-common/cadence-samples) to get started.

Documentation is available [here](https://cadenceworkflow.io/docs/go-client/).
You can also find the API documentation [here](https://godoc.org/go.uber.org/cadence).

## Contributing
We'd love your help in making the Cadence Go client great. Please review our [contribution guidelines](CONTRIBUTING.md).

If you'd like to propose a new feature or discuss issues, join the [CNCF Slack workspace](https://communityinviter.com/apps/cloud-native/cncf) in the **#cadence-users** channel to start a discussion.

## License
Apache 2.0 License, please see [LICENSE](LICENSE) for details.


Cadence-client

A framework developed in Go that manages the execution of workflows described by directed acyclic graphs

# go-dag
A framework developed in Go that manages the execution of workflows described by directed acyclic graphs.

[![Go Reference](https://pkg.go.dev/badge/github.com/rhosocial/go-dag.svg)](https://pkg.go.dev/github.com/rhosocial/go-dag)
[![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go)  
![GitHub Tag](https://img.shields.io/github/v/tag/rhosocial/go-dag)
![GitHub go.mod Go version](https://img.shields.io/github/go-mod/go-version/rhosocial/go-dag)
[![Go Report Card](https://goreportcard.com/badge/github.com/rhosocial/go-dag)](https://goreportcard.com/report/github.com/rhosocial/go-dag)
![GitHub Actions Workflow Status](https://img.shields.io/github/actions/workflow/status/rhosocial/go-dag/go.yml?branch=r1.0)
[![Codecov coverage](https://codecov.io/gh/rhosocial/go-dag/branch/r1.0/graph/badge.svg)](https://app.codecov.io/gh/rhosocial/go-dag/tree/r1.0)

Unlike any other DAG framework, `GO-DAG` does not emphasize concepts such as edges and vertices,
but allows users to focus on defining `transit`(worker and its inputs and outputs).
`GO-DAG` will automatically builds workflows based on user-defined transits and then executes them.

Now this project is fully tested in Go 1.20, 1.21, 1.22, and always be adapted to the latest version.

This project always adheres to zero external dependencies,
that is, it only relies on internal code packages (except for testing).

## Introduction

`GO-DAG` consists of several code packages that differ in the complexity of the described workflows.

Currently, the released versions include:

- [workflow/simple](workflow/simple): used to describe simple workflows, supports DAGs of any complexity,
and has logging capabilities that allow custom transits and log events. Although simple, it can satisfy most scenarios.

> In the future, workflows with more complex functions will be introduced.

## Reference

If you want to know the specific usage of `GO-DAG` and more details, please visit [this](https://docs.go-dag.dev.rho.social/).

If you want real-person online guidance on how to use it, you can:

[![Gitter](https://img.shields.io/gitter/room/rhosocial/go-dag)](https://matrix.to/#/#go-dag.rhosocial:gitter.im)

If you want guidance on using this framework to transform your complex tasks into the most efficient workflows, you can start by donating to:

[![Donate](https://liberapay.com/assets/widgets/donate.svg)](https://liberapay.com/vistart/donate)


The Best Go Libraries For Workflow Frameworks (5)

Dagu

go-taskflow

workflow

Cadence-client

go-dag