durabletask

package
v0.0.0-...-4f198ce Latest Latest
Warning

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

 Go to latest Published: Oct 2, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package durabletask provides a replicadb-backed implementation of the Durable Task Framework backend interfaces.

This backend persists orchestration and activity state using ReplicaDB and implements the backend.Backend interface from durabletask-go.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func NoopLogger 

func NoopLogger() backend.Logger

NoopLogger is a backend logger that logs nothing.

func WithActivityLockTimeout 

func WithActivityLockTimeout(timeout time.Duration) func(b *ReplicaDBBackend)

WithActivityLockTimeout configures the activity lock timeout. A worker renews the lock of each activity it runs every third of the timeout, so an activity may run for as long as it needs; the timeout is how soon a crashed worker's activities pass to another worker.

func WithLogger 

func WithLogger(logger backend.Logger) func(b *ReplicaDBBackend)

WithLogger configures the logger used by the backend.

func WithMaxParallelism 

func WithMaxParallelism(n int32) func(b *ReplicaDBBackend)

WithMaxParallelism configures how many work items each of an executor's workers, orchestration and activity, runs at once. The default is one.

func WithOrchestrationLockTimeout 

func WithOrchestrationLockTimeout(timeout time.Duration) func(b *ReplicaDBBackend)

WithOrchestrationLockTimeout configures the orchestration lock timeout.

Types

type Executor 

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

func NewExecutor 

func NewExecutor(db *replicadb.DB, options ...Option) *Executor

func (*Executor) EnableCompactor 

func (e *Executor) EnableCompactor(interval time.Duration) *Executor

func (*Executor) ExternalEvent 

func (e *Executor) ExternalEvent(ctx context.Context, id api.InstanceID, eventName string, opts ...api.RaiseEventOptions) error

func (*Executor) OrchestrationMetadata 

func (e *Executor) OrchestrationMetadata(ctx context.Context, id api.InstanceID) (*api.OrchestrationMetadata, error)

func (*Executor) OrchestrationsWithName 

func (e *Executor) OrchestrationsWithName(ctx context.Context, name string) ([]*api.OrchestrationMetadata, error)

func (*Executor) PurgeCompletedOrchestrationsOlderThan 

func (e *Executor) PurgeCompletedOrchestrationsOlderThan(ctx context.Context, timeframe time.Duration, opts ...api.PurgeOptions) error

func (*Executor) PurgeOrchestration 

func (e *Executor) PurgeOrchestration(ctx context.Context, id api.InstanceID, opts ...api.PurgeOptions) error

func (*Executor) RegisterActivity 

func (e *Executor) RegisterActivity(name string, activity task.Activity) error

func (*Executor) RegisterOrchestration 

func (e *Executor) RegisterOrchestration(name string, orchestrator task.Orchestrator) error

func (*Executor) ResumeOrchestration 

func (e *Executor) ResumeOrchestration(ctx context.Context, id api.InstanceID, reason string) error

func (*Executor) ScheduleOrchestration 

func (e *Executor) ScheduleOrchestration(ctx context.Context, name string, opts ...api.NewOrchestrationOptions) (api.InstanceID, error)

func (*Executor) Shutdown 

func (e *Executor) Shutdown(ctx context.Context) error

Shutdown stops the workers' pollers, and once each has stopped, drains the workers.

NB: durabletask-go's worker adds to its pending WaitGroup each time it polls, and draining waits on that group, so draining while a poller may poll again races. Stopping the pollers first, and waiting for each to say it has, keeps the two apart.

func (*Executor) Start 

func (e *Executor) Start(ctx context.Context) error

func (*Executor) SuspendOrchestration 

func (e *Executor) SuspendOrchestration(ctx context.Context, id api.InstanceID, reason string) error

func (*Executor) TerminateOrchestration 

func (e *Executor) TerminateOrchestration(ctx context.Context, id api.InstanceID, opts ...api.TerminateOptions) error

func (*Executor) WaitForOrchestrationCompletion 

func (e *Executor) WaitForOrchestrationCompletion(ctx context.Context, id api.InstanceID) (*api.OrchestrationMetadata, error)

func (*Executor) WaitForOrchestrationStart 

func (e *Executor) WaitForOrchestrationStart(ctx context.Context, id api.InstanceID) (*api.OrchestrationMetadata, error)

type Option 

type Option func(b *ReplicaDBBackend)

Option represents a functional option for configuring a ReplicaDBBackend.

type ReplicaDBBackend 

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

ReplicaDBBackend implements the durabletask-go backend.Backend interface using replicadb as the persistence layer.

func NewReplicaDBBackend 

func NewReplicaDBBackend(db *replicadb.DB, options ...Option) *ReplicaDBBackend

NewReplicaDBBackend creates a new replicadb-backed durable task backend.

The returned value implements backend.Backend and is ready to be started after applying any provided configuration options.

func (*ReplicaDBBackend) AbandonActivityWorkItem 

func (b *ReplicaDBBackend) AbandonActivityWorkItem(ctx context.Context, wi *backend.ActivityWorkItem) error

AbandonActivityWorkItem releases the lock on an activity task, making it available for retry by another worker.

func (*ReplicaDBBackend) AbandonOrchestrationWorkItem 

func (b *ReplicaDBBackend) AbandonOrchestrationWorkItem(ctx context.Context, wi *backend.OrchestrationWorkItem) error

AbandonOrchestrationWorkItem releases locks held on an orchestration and optionally delays re-visibility of its events.

func (*ReplicaDBBackend) AddNewOrchestrationEvent 

func (b *ReplicaDBBackend) AddNewOrchestrationEvent(ctx context.Context, iid api.InstanceID, e *backend.HistoryEvent) error

AddNewOrchestrationEvent enqueues a new external event for an existing orchestration instance.

func (*ReplicaDBBackend) CompleteActivityWorkItem 

func (b *ReplicaDBBackend) CompleteActivityWorkItem(ctx context.Context, wi *backend.ActivityWorkItem) error

CompleteActivityWorkItem completes an activity task by:

  • Enqueuing its result as a new orchestration event
  • Deleting the activity task if the lock is still valid

func (*ReplicaDBBackend) CompleteOrchestrationWorkItem 

func (b *ReplicaDBBackend) CompleteOrchestrationWorkItem(ctx context.Context, wi *backend.OrchestrationWorkItem) error

CompleteOrchestrationWorkItem finalizes processing of an orchestration work item.

This method:

  • Updates orchestration instance metadata (status, timestamps, output)
  • Appends new history events
  • Enqueues outbound activity tasks and orchestration messages
  • Deletes processed inbound events

All operations are performed atomically within a transaction. If the orchestration lock is lost, ErrWorkItemLockLost is returned.

func (*ReplicaDBBackend) CreateOrchestrationInstance 

func (b *ReplicaDBBackend) CreateOrchestrationInstance(ctx context.Context, e *backend.HistoryEvent, opts ...backend.OrchestrationIdReusePolicyOptions) error

CreateOrchestrationInstance creates a new orchestration instance and enqueues its ExecutionStarted event for processing.

If an instance with the same ID already exists, the configured reuse policy determines whether the request is ignored, rejected, or replaces the existing instance.

func (*ReplicaDBBackend) CreateTaskHub 

func (b *ReplicaDBBackend) CreateTaskHub(context.Context) error

CreateTaskHub creates the task hub schema in the underlying database.

func (*ReplicaDBBackend) DeleteTaskHub 

func (b *ReplicaDBBackend) DeleteTaskHub(ctx context.Context) error

DeleteTaskHub deletes all task hub tables from the database.

func (*ReplicaDBBackend) GetActivityWorkItem 

func (b *ReplicaDBBackend) GetActivityWorkItem(ctx context.Context) (*backend.ActivityWorkItem, error)

GetActivityWorkItem acquires and locks the next available activity task for execution.

If no activity work is available, ErrNoWorkItems is returned.

func (*ReplicaDBBackend) GetOrchestrationMetadata 

func (b *ReplicaDBBackend) GetOrchestrationMetadata(ctx context.Context, iid api.InstanceID) (*api.OrchestrationMetadata, error)

GetOrchestrationMetadata retrieves metadata for a specific orchestration instance, including runtime status, timestamps, input/output, and failure details.

func (*ReplicaDBBackend) GetOrchestrationRuntimeState 

func (b *ReplicaDBBackend) GetOrchestrationRuntimeState(ctx context.Context, wi *backend.OrchestrationWorkItem) (*backend.OrchestrationRuntimeState, error)

GetOrchestrationRuntimeState loads the full orchestration runtime state, including all historical events, for replay by the orchestrator.

func (*ReplicaDBBackend) GetOrchestrationWorkItem 

func (b *ReplicaDBBackend) GetOrchestrationWorkItem(ctx context.Context) (*backend.OrchestrationWorkItem, error)

GetOrchestrationWorkItem acquires a lock on an orchestration instance that has visible, unprocessed events and returns it for execution.

If no work is available, ErrNoWorkItems is returned.

func (*ReplicaDBBackend) GetOrchestrationsWithName 

func (b *ReplicaDBBackend) GetOrchestrationsWithName(ctx context.Context, name string) ([]*api.OrchestrationMetadata, error)

GetOrchestrationsWithName returns orchestration metadata of orchestrations with the given status.

func (*ReplicaDBBackend) GetOrchestrationsWithStatus 

func (b *ReplicaDBBackend) GetOrchestrationsWithStatus(ctx context.Context, status api.OrchestrationStatus) ([]*api.OrchestrationMetadata, error)

GetOrchestrationsWithStatus returns orchestration metadata of orchestrations with the given status.

func (*ReplicaDBBackend) PurgeCompletedOrchestrationStateOlderThan 

func (b *ReplicaDBBackend) PurgeCompletedOrchestrationStateOlderThan(ctx context.Context, timeframe time.Duration, opts ...api.PurgeOptions) error

PurgeCompletedOrchestrationStateBefore permanently deletes all persisted state for all orchestration instances, provided they have completed prior to the given time.

func (*ReplicaDBBackend) PurgeOrchestrationState 

func (b *ReplicaDBBackend) PurgeOrchestrationState(ctx context.Context, id api.InstanceID) error

PurgeOrchestrationState permanently deletes all persisted state for an orchestration instance, provided it has completed.

func (*ReplicaDBBackend) Start 

func (*ReplicaDBBackend) Start(context.Context) error

Start initializes the backend.

ReplicaDBBackend does not require explicit startup logic.

func (*ReplicaDBBackend) Stop 

func (b *ReplicaDBBackend) Stop(context.Context) error

Stop shuts down the backend, and stops renewing the locks it holds.

Source Files

View all Source files

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.