durabletask-go

Durable Task Framework for Go

[!IMPORTANT] This repository is an independently maintained, hardened fork of Durable Task for Go. Its lineage is Microsoft Durable Task for Go, then Dapr Durable Task for Go, then this fork. It is maintained by Mafilus and is not affiliated with, endorsed by, or supported by Microsoft or the Dapr project.

Its primary goal is to freeze and stabilize API and runtime behavior for production use. Changes favor compatibility, correctness, and operational safety within this fork over upstream feature parity.

The Durable Task Framework is a lightweight, embeddable engine for writing durable, fault-tolerant business logic (orchestrations) as ordinary code. The engine itself is written in Go and intended to be embedded into other Go-based processes. It exposes a gRPC endpoint to support writing durable flows in any language. There are currently SDKs that consume this gRPC endpoint for .NET and Java, with more to come. It's also possible to write orchestrations directly in Go and run them in the local process.

The project originated as a Go clone of the .NET-based Durable Task Framework. It also takes inspiration from the Go Workflows project and Temporal. This fork is designed to be embedded in Go-based processes and sidecar-style deployments; it does not follow the product roadmap or contribution processes of its upstream projects.

Current release

The current stable release is v1.2.1. It requires Go 1.26.6 or newer, as declared in go.mod.

go get github.com/mafilus/durabletask-go@v1.2.1

See the release note for upgrade instructions and validation evidence, including the separately triggered PostgreSQL chaos run.

Compatibility posture

The public contract of this fork is defined by its own Go module and release tags. Compatibility with Dapr's workflow Actors backend, its protobuf feature surface, or its persisted workflow histories is not a goal of this project.

Version v1.0.0 removes RerunWorkflowFromEvent from the Go and gRPC APIs. The operation was never implemented by the bundled SQLite or PostgreSQL backends and its complete implementation is specific to Dapr Actors. A generic workflow engine must start a new workflow instance when an application wants a new execution. See the v1 migration note before upgrading.

Version v1.1.0 adds opt-in, durable deduplication for at-least-once external events. PostgreSQL operators must apply its schema migration before enabling the option; see the v1.1.0 release note.

Version v1.1.1 fixes workflow termination and execution identity without an API or schema change; see the maintenance release note.

Version v1.2.0 adds early-resolution handling and worker transport recovery. Upgrade workers before enabling backend.WithRequireCompletionTokens() on the server: legacy responses without a token remain accepted by default. See the v1.2.0 release note for the compatibility and proof boundaries.

Version v1.2.1 fixes execution isolation across cancellation and recreation, and updates gRPC for two security fixes. Before deploying it, coordinate SQL writers and drain or migrate ambiguous legacy child results as described in the v1.2.1 release note.

Security responsibility boundary

Durabletask-go is a neutral workflow engine. Historical signatures, identity attestation, and integrity-policy enforcement are not responsibilities of this repository.

An integrating runtime is responsible for deciding and enforcing any required identity, authorization, integrity, provenance, or audit policy around workflow execution and durable storage.

Storage providers

This project includes a sqlite storage provider for persisting app state to disk.

// Persists state to a file named test.sqlite3. Use "" for in-memory storage.
options := sqlite.NewSqliteOptions("test.sqlite3")
be := sqlite.NewSqliteBackend(options, backend.DefaultLogger())

Additional storage providers can be created by extending the Backend interface.

Creating the standalone gRPC sidecar

See the main.go file for an example of how to create a standalone gRPC sidecar that embeds the Durable Task engine. In short, you must create an Backend (for storage), an Executor (for executing user code), and host them as a TaskHubWorker.

The following code creates a TaskHub worker with sqlite Backend and a gRPC Executor implementations.

// Use the default logger or provide your own
logger := backend.DefaultLogger()

// Configure the sqlite backend that will store the runtime state
sqliteOptions := sqlite.NewSqliteOptions(sqliteFilePath)
be := sqlite.NewSqliteBackend(sqliteOptions, logger)

// Create a gRPC server that the language SDKs will connect to
grpcServer := grpc.NewServer()
executor, registerFn := backend.NewGrpcExecutor(be, logger)
registerFn(grpcServer)

// Construct and start the task hub worker object, which polls the backend for new work
workflowWorker := backend.NewWorkflowWorker(backend.WorkflowWorkerOptions{
	Backend:  be,
	Executor: executor,
	Logger:   logger,
	AppID:    "example",
})
activityWorker := backend.NewActivityTaskWorker(be, executor, logger)
taskHubWorker := backend.NewTaskHubWorker(be, workflowWorker, activityWorker, logger)
taskHubWorker.Start(context.Background())

// Start listening.
lis, _ := net.Listen("tcp", "localhost:4001")
fmt.Printf("server listening at %v\n", lis.Addr())
grpcServer.Serve(lis)

Note that the Durable Task gRPC service implementation is designed to serve one client at a time, just like with any sidecar architecture. Scale out is achieved by adding new pod replicas that contain both the app process and the sidecar (connected to a common database).

Language SDKs for gRPC

The Durable Task Framework for Go currently supports writing orchestrations in the following languages:

Language/Stack	Package	Project Home	Samples
.NET		GitHub	Samples
Java		GitHub	Samples
Python		GitHub	Samples

More language SDKs are planned to be added in the future. In particular, SDKs for Python and JavaScript/TypeScript. Anyone can theoretically create an SDK using a language that supports gRPC. However, there is not yet a guide for how to do this, so developers would need to reference existing SDK code as a reference. Starting with the Java implementation is recommended. The gRPC API is defined here.

Embedded orchestrations

It's also possible to create orchestrations in Go and run them in the local process. The full set of Durable Task features is not yet available as part of the Go SDK, but will be added over time.

You can find code samples in the samples directory.
To run them, get into the folder of each sample and run go run .

Activity sequence example

Activity sequences like the following are the simplest and most common pattern used in the Durable Task Framework.

// ActivitySequenceOrchestrator makes three activity calls in sequence and results the results
// as an array.
func ActivitySequenceOrchestrator(ctx *task.OrchestrationContext) (any, error) {
	var helloTokyo string
	if err := ctx.CallActivity(SayHelloActivity, task.WithActivityInput("Tokyo")).Await(&helloTokyo); err != nil {
		return nil, err
	}
	var helloLondon string
	if err := ctx.CallActivity(SayHelloActivity, task.WithActivityInput("London")).Await(&helloLondon); err != nil {
		return nil, err
	}
	var helloSeattle string
	if err := ctx.CallActivity(SayHelloActivity, task.WithActivityInput("Seattle")).Await(&helloSeattle); err != nil {
		return nil, err
	}
	return []string{helloTokyo, helloLondon, helloSeattle}, nil
}

// SayHelloActivity can be called by an orchestrator function and will return a friendly greeting.
func SayHelloActivity(ctx task.ActivityContext) (any, error) {
	var input string
	if err := ctx.GetInput(&input); err != nil {
		return "", err
	}
	return fmt.Sprintf("Hello, %s!", input), nil
}

You can find the full sample here.

Fan-out / fan-in execution example

The next most common pattern is "fan-out / fan-in" where multiple activities are run in parallel, as shown in the snippet below (note that the GetDevicesToUpdate and UpdateDevice activity definitions are left out of the snippet below for brevity):

// UpdateDevicesOrchestrator is an orchestrator that runs activities in parallel
func UpdateDevicesOrchestrator(ctx *task.OrchestrationContext) (any, error) {
	// Get a dynamic list of devices to perform updates on
	var devices []string
	if err := ctx.CallActivity(GetDevicesToUpdate).Await(&devices); err != nil {
		return nil, err
	}

	// Start a dynamic number of tasks in parallel, not waiting for any to complete (yet)
	tasks := make([]task.Task, len(devices))
	for i, id := range devices {
		tasks[i] = ctx.CallActivity(UpdateDevice, task.WithActivityInput(id))
	}

	// Now that all are started, wait for them to complete and then return the success rate
	successCount := 0
	for _, task := range tasks {
		var succeeded bool
		if err := task.Await(&succeeded); err == nil && succeeded {
			successCount++
		}
	}

	return float32(successCount) / float32(len(devices)), nil
}

The full sample can be found here.

External orchestration inputs (events) example

Sometimes orchestrations need asynchronous input from external systems. For example, an approval workflow may require a manual approval signal from an authorized user. Or perhaps an orchestration pauses and waits for a command from an operator. The WaitForSingleEvent method can be used in an orchestrator function to pause execution and wait for such inputs. You an even specify a timeout value indicating how long to wait for the input before resuming execution (use -1 to indicate infinite timeout).

// ExternalEventOrchestrator is an orchestrator function that blocks for 30 seconds or
// until a "Name" event is sent to it.
func ExternalEventOrchestrator(ctx *task.OrchestrationContext) (any, error) {
	var nameInput string
	if err := ctx.WaitForSingleEvent("Name", 30*time.Second).Await(&nameInput); err != nil {
		// Timeout expired
		return nil, err
	}

	return fmt.Sprintf("Hello, %s!", nameInput), nil
}

Sending an event to a waiting orchestration can be done using the RaiseEvent method of the task hub client. These events are durably buffered in the orchestration state and are consumed as soon as the target orchestration calls WaitForSingleEvent with a matching event name. The following code shows how to use the RaiseEvent method to send an event with a payload to a running orchestration. See Managing local orchestrations for more information on how to interact with local orchestrations in Go.

id, _ := client.ScheduleNewOrchestration(ctx, ExternalEventOrchestrator)

// Prompt the user for their name and send that to the orchestrator
go func() {
	fmt.Println("Enter your first name: ")
	var nameInput string
	fmt.Scanln(&nameInput)
	
	client.RaiseEvent(ctx, id, "Name", api.WithEventPayload(nameInput))
}()

The full sample can be found here.

Managing local orchestrations

The following code snippet provides an example of how you can configure and run orchestrations. The TaskRegistry type allows you to register orchestrator and activity functions, and the TaskHubClient allows you to start, query, terminate, suspend, resume, and wait for orchestrations to complete.

The code snippet below demonstrates how to register and start a new instance of the ActivitySequenceOrchestrator orchestrator and wait for it to complete. The initialization of the client and worker are left out for brevity.

r := task.NewTaskRegistry()
r.AddOrchestrator(ActivitySequenceOrchestrator)
r.AddActivity(SayHelloActivity)

ctx := context.Background()
client, worker := Init(ctx, r)
defer worker.Shutdown(ctx)

id, err := client.ScheduleNewOrchestration(ctx, ActivitySequenceOrchestrator)
if err != nil {
  panic(err)
}

metadata, err := client.WaitForOrchestrationCompletion(ctx, id)
if err != nil {
  panic(err)
}

fmt.Printf("orchestration completed: %v\n", metadata)

Each sample linked above has a full implementation you can use as a reference.

Distributed tracing support

The Durable Task Framework for Go supports publishing distributed traces to any configured OpenTelemetry-compatible exporter. Simply use otel.SetTracerProvider(tp) to register a global TracerProvider as part of your application startup and the task hub worker will automatically use it to emit OTLP trace spans.

The following example configures distributed trace collection through OTLP/HTTP. It assumes an OpenTelemetry Collector is listening locally on port 4318.

func ConfigureOTLPTracing(ctx context.Context) (*trace.TracerProvider, error) {
	exp, err := otlptracehttp.New(ctx,
		otlptracehttp.WithEndpoint("localhost:4318"),
		otlptracehttp.WithInsecure(),
	)
	if err != nil {
		return nil, err
	}

	processor := trace.NewBatchSpanProcessor(exp)

	tp := trace.NewTracerProvider(
		trace.WithSpanProcessor(processor),
		trace.WithSampler(trace.AlwaysSample()),
		trace.WithResource(resource.NewWithAttributes(
			"durabletask.io",
			attribute.KeyValue{Key: "service.name", Value: attribute.StringValue("sample-app")},
		)),
	)
	otel.SetTracerProvider(tp)
	return tp, nil
}

You can find this code in the distributedtracing sample.

Example trace

The sample orchestration produces a distributed trace similar to the following:

The complete trace contains 6 spans, has a maximum depth of 4, and completes in approximately 3.5 seconds.

The orchestration span exposes Durable Task metadata through OpenTelemetry attributes, including:

Note that each orchestration is represented as a single span with activities, timers, and sub-orchestrations as child spans. The generated spans contain a variety of attributes that include information such as orchestration instance IDs, task names, task IDs, etc.

Running integration tests

You can run pre-built container images to run full integration tests against the durable task host over gRPC.

.NET Durable Task client SDK tests

Use the following docker command to run tests against a running worker.

docker run -e GRPC_HOST="host.docker.internal" cgillum/durabletask-dotnet-tester:0.5.0-beta

Note that the test assumes the gRPC server can be reached over localhost on port 4001 on the host machine. These values can be overridden with the following environment variables:

• GRPC_HOST: Use this to change from the default 127.0.0.1 to some other value, for example host.docker.internal.
• GRPC_PORT: Set this environment variable to change the default port from 4001 to something else.

If successful, you should see output that looks like the following:

Test run for /root/out/bin/Debug/Microsoft.DurableTask.Tests/net6.0/Microsoft.DurableTask.Tests.dll (.NETCoreApp,Version=v6.0)
Microsoft (R) Test Execution Command Line Tool Version 17.3.1 (x64)
Copyright (c) Microsoft Corporation.  All rights reserved.

Starting test execution, please wait...
A total of 1 test files matched the specified pattern.
[xUnit.net 00:00:00.00] xUnit.net VSTest Adapter v2.4.3+1b45f5407b (64-bit .NET 6.0.10)
[xUnit.net 00:00:00.82]   Discovering: Microsoft.DurableTask.Tests
[xUnit.net 00:00:00.90]   Discovered:  Microsoft.DurableTask.Tests
[xUnit.net 00:00:00.90]   Starting:    Microsoft.DurableTask.Tests
  Passed Microsoft.DurableTask.Tests.OrchestrationPatterns.ExternalEvents(eventCount: 100) [6 s]
  Passed Microsoft.DurableTask.Tests.OrchestrationPatterns.ExternalEvents(eventCount: 1) [309 ms]
  Passed Microsoft.DurableTask.Tests.OrchestrationPatterns.LongTimer [8 s]
  Passed Microsoft.DurableTask.Tests.OrchestrationPatterns.SubOrchestration [1 s]
  ...
  Passed Microsoft.DurableTask.Tests.OrchestrationPatterns.ActivityFanOut [914 ms]
[xUnit.net 00:01:01.04]   Finished:    Microsoft.DurableTask.Tests
  Passed Microsoft.DurableTask.Tests.OrchestrationPatterns.SingleActivity_Async [365 ms]

Test Run Successful.
Total tests: 33
     Passed: 33
 Total time: 1.0290 Minutes

Running locally

You can run the engine locally by pressing F5 in Visual Studio Code (the recommended editor). You can also simply run go run main.go to start a local Durable Task gRPC server that listens on port 4001.

go run main.go --port 4001 --db ./test.sqlite3

The following is the expected output:

2022/09/14 17:26:50 backend started: sqlite::./test.sqlite3
2022/09/14 17:26:50 server listening at 127.0.0.1:4001
2022/09/14 17:26:50 orchestration-processor: waiting for new work items...
2022/09/14 17:26:50 activity-processor: waiting for new work items...

At this point you can use one of the language SDKs mentioned earlier in a separate process to implement and execute durable orchestrations. Those SDKs will connect to port 4001 by default to interact with the Durable Task engine.

gRPC listener security

The server and container image bind to loopback (127.0.0.1) by default. Plaintext gRPC is allowed only for a loopback listener, intended for local development or a trusted local proxy.

To bind to a non-loopback address, configure mTLS and an allowlist of client certificate URI SANs. The server rejects startup otherwise:

export DURABLETASK_GRPC_HOST=0.0.0.0
export DURABLETASK_GRPC_TLS_CERT_FILE=/run/tls/server.crt
export DURABLETASK_GRPC_TLS_KEY_FILE=/run/tls/server.key
export DURABLETASK_GRPC_CLIENT_CA_FILE=/run/tls/client-ca.crt
export DURABLETASK_GRPC_ALLOWED_CLIENT_URIS=urn:durabletask:client

DURABLETASK_GRPC_PORT overrides the listener port. When exposing a container port, also apply a network policy that limits clients to the intended identities and network paths.

Contributing

Contributions and suggestions are welcome through this repository's issues and pull requests. For substantial changes, open an issue first so that scope and compatibility can be discussed.

By submitting a contribution, you confirm that you have the right to submit it and agree that it is provided under the repository's Apache-2.0 license. No Microsoft or Dapr contributor-license agreement, bot, or governance process applies to this fork.

Please follow the repository's Code of Conduct.

License and provenance

This repository is distributed under Apache-2.0. Applicable copyright, license, and attribution notices from Microsoft, Dapr, and other upstream contributors are retained. Copyright notices for changes made in this fork identify their respective authors; they do not imply ownership of the entire repository.

Trademarks

Third-party names and marks are used only to describe this repository's technical history and compatibility. They remain the property of their respective owners and do not imply affiliation with, endorsement of, or support for this fork.