workflow

package
v0.0.17 Latest Latest
Warning

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

 Go to latest Published: Jul 28, 2026 License: Apache-2.0 Imports: 21 Imported by: 0

Documentation

Index

Constants

View Source 
const WorkflowComponent = "dapr"

Variables

View Source 
var (
	ErrNotFound         = errors.New("workflow not found")
	ErrNoStore          = errors.New("no state store configured")
	ErrStoreUnreachable = errors.New("could not connect to state store")
)
View Source 
var ErrSidecarUnsupported = errors.New("workflow inspection via the sidecar requires Dapr 1.17 or newer")

ErrSidecarUnsupported marks a sidecar whose runtime predates the workflow management API (ListInstanceIDs/GetInstanceHistory, Dapr 1.17+).

Functions

This section is empty.

Types

type EndpointsFunc added in v0.0.10 

type EndpointsFunc func(ctx context.Context) []SidecarEndpoint

EndpointsFunc returns the current set of sidecar-sourced apps. It is called per query so discovery changes (new published ports after a test rerun) apply immediately.

type Execution 

type Execution struct {
	ExecutionSummary
	Input          *string         `json:"input,omitempty"`
	Output         *string         `json:"output,omitempty"`
	CustomStatus   string          `json:"customStatus,omitempty"`
	ReplayCount    int             `json:"replayCount"`
	FailureDetails *FailureDetails `json:"failureDetails,omitempty"`
	History        []HistoryEvent  `json:"history"`
}

func DecodeExecution 

func DecodeExecution(appID, instanceID string, history []*protos.HistoryEvent, customStatus string) Execution

DecodeExecution builds a full Execution from an instance's history events.

type ExecutionSummary 

type ExecutionSummary struct {
	AppID            string     `json:"appId"`
	InstanceID       string     `json:"instanceId"`
	Name             string     `json:"name"`
	Status           Status     `json:"status"`
	ParentInstanceID string     `json:"parentInstanceId,omitempty"` // non-empty ⇒ this is a child workflow
	CreatedAt        *time.Time `json:"createdAt,omitempty"`
	LastUpdatedAt    *time.Time `json:"lastUpdatedAt,omitempty"`
}

type FailureDetails 

type FailureDetails struct {
	ErrorType  string `json:"errorType,omitempty"`
	Message    string `json:"message,omitempty"`
	StackTrace string `json:"stackTrace,omitempty"`
}

type HistoryEvent 

type HistoryEvent struct {
	SequenceID     int32           `json:"sequenceId"`
	Timestamp      time.Time       `json:"timestamp"`
	Type           string          `json:"type"`
	Name           string          `json:"name,omitempty"`
	InstanceID     string          `json:"instanceId,omitempty"`  // child instance id for SubOrchestrationCreated
	ScheduledID    *int32          `json:"scheduledId,omitempty"` // start event's EventId; set on completion/fired events
	Input          *string         `json:"input,omitempty"`
	Output         *string         `json:"output,omitempty"`
	FailureDetails *FailureDetails `json:"failureDetails,omitempty"` // set on TaskFailed / SubOrchestrationFailed / ExecutionFailed
}

type ListQuery 

type ListQuery struct {
	AppID           string
	Status          []Status
	Search          string
	PageSize        int
	PageToken       string
	IncludeChildren bool
}

type ListResult 

type ListResult struct {
	Items     []ExecutionSummary `json:"items"`
	NextToken string             `json:"nextToken,omitempty"`
}

ListResult is one page of executions plus the cursor for the next page.

Limitations of the underlying key-cursor paging:

  • Items are sorted by CreatedAt (newest first) within this page only; the store pages by key order, so there is no global CreatedAt order across pages.
  • A page may hold fewer than the requested pageSize items — down to zero when a filtered scan hits its per-request cap — while NextToken is still non-empty. Clients must keep paging until NextToken is empty rather than treating a short or empty page as the end.

type Mechanism 

type Mechanism string
const (
	MechTerminateThenPurge Mechanism = "terminate_then_purge"
	MechPurge              Mechanism = "purge"
	MechForce              Mechanism = "force"
)

func SelectMechanism 

func SelectMechanism(status Status, sidecarHealthy, force bool) Mechanism

SelectMechanism chooses the removal path for one workflow (spec §7).

type Option added in v0.0.9 

type Option func(*service)

Option customizes a workflow Service.

func WithNamespaceResolver added in v0.0.9 

func WithNamespaceResolver(fn func(ctx context.Context, appID string) string) Option

WithNamespaceResolver injects a per-app namespace lookup used by app-scoped operations. fn returns "" to fall back to the store namespace.

type RemoveResult 

type RemoveResult struct {
	InstanceID string    `json:"instanceId"`
	Mechanism  Mechanism `json:"mechanism"`
	OK         bool      `json:"ok"`
	Error      string    `json:"error,omitempty"`
}

type RemoveTarget 

type RemoveTarget struct {
	AppID      string
	InstanceID string
	Status     Status
	HTTPPort   int
	// DaprHTTPBaseURL, when set (aspire-discovered apps), replaces
	// http://127.0.0.1:<HTTPPort> for the terminate/purge calls.
	DaprHTTPBaseURL string
	Healthy         bool
	// Namespace, when non-empty (aspire per-app DEVDASHBOARD_APP_<i>_NAMESPACE),
	// scopes the force-delete state-key pattern to the app's own namespace.
	// Empty (host-scanned apps) falls back to the remover's store namespace.
	Namespace string
}

type Remover 

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

func NewRemover 

func NewRemover(client *http.Client, store statestore.Store, namespace string) *Remover

func (*Remover) Remove 

func (r *Remover) Remove(ctx context.Context, t RemoveTarget, force bool) RemoveResult

func (*Remover) RemoveMany 

func (r *Remover) RemoveMany(ctx context.Context, targets []RemoveTarget, force bool) []RemoveResult

type Service 

type Service interface {
	List(ctx context.Context, q ListQuery) (ListResult, error)
	Stats(ctx context.Context, q ListQuery) (StatsResult, error)
	Get(ctx context.Context, appID, instanceID string) (Execution, error)
	// AppIDs returns every distinct app-id that has workflow data in the store,
	// independent of any list filter — the source of truth for the app filter.
	AppIDs(ctx context.Context) ([]string, error)
}

func New 

func New(store statestore.Store, namespace string, opts ...Option) Service

func NewComposite added in v0.0.10 

func NewComposite(base Service, sc *SidecarService) Service

NewComposite builds the per-app routing service. base is the store-backed service (possibly degraded/unreachable); sc is the sidecar source.

func NewUnreachableService 

func NewUnreachableService(name, conn string) Service

NewUnreachableService builds a Service whose List/Stats/Get all fail with a store-specific ErrStoreUnreachable error.

type SidecarEndpoint added in v0.0.10 

type SidecarEndpoint struct {
	AppID string
	Addr  string // host:port, e.g. "127.0.0.1:58445"
}

SidecarEndpoint is one app's daprd gRPC endpoint.

type SidecarPool added in v0.0.10 

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

SidecarPool caches one gRPC client connection per endpoint address. grpc.NewClient is lazy, so entries are cheap; connections to ports from finished test runs stay idle until Close. Close on shutdown.

func NewSidecarPool added in v0.0.10 

func NewSidecarPool() *SidecarPool

func (*SidecarPool) Close added in v0.0.10 

func (p *SidecarPool) Close() error

func (*SidecarPool) Service added in v0.0.10 

func (p *SidecarPool) Service(eps EndpointsFunc) *SidecarService

Service returns a workflow Service reading from the sidecars selected by eps.

type SidecarService added in v0.0.10 

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

SidecarService reads workflow data live from daprd's gRPC workflow management API instead of the state store. It works with any backing store — state.in-memory included — because the sidecar itself answers.

func (*SidecarService) AppIDs added in v0.0.10 

func (s *SidecarService) AppIDs(ctx context.Context) ([]string, error)

AppIDs returns the sidecar-sourced apps that hold at least one instance.

func (*SidecarService) Get added in v0.0.10 

func (s *SidecarService) Get(ctx context.Context, appID, instanceID string) (Execution, error)

func (*SidecarService) HasEndpoints added in v0.0.10 

func (s *SidecarService) HasEndpoints(ctx context.Context) bool

HasEndpoints reports whether any app is currently sidecar-sourced.

func (*SidecarService) List added in v0.0.10 

func (s *SidecarService) List(ctx context.Context, q ListQuery) (ListResult, error)

List returns every matching instance across the sidecar-sourced apps in one page (NextToken is always empty: local sidecars hold bounded instance counts, so store-style cursor paging buys nothing here). A failing app is skipped and logged — one down sidecar never empties the whole page.

func (*SidecarService) Owns added in v0.0.10 

func (s *SidecarService) Owns(ctx context.Context, appID string) bool

Owns reports whether appID is served by this sidecar source.

func (*SidecarService) Stats added in v0.0.10 

func (s *SidecarService) Stats(ctx context.Context, q ListQuery) (StatsResult, error)

Stats tallies statuses across all matching instances (Status filter and paging ignored, mirroring the store-backed Stats contract).

type StatsResult 

type StatsResult struct {
	Counts map[Status]int `json:"counts"`
	Total  int            `json:"total"`
}

type Status 

type Status string
const (
	StatusPending    Status = "Pending"
	StatusRunning    Status = "Running"
	StatusCompleted  Status = "Completed"
	StatusFailed     Status = "Failed"
	StatusTerminated Status = "Terminated"
	StatusSuspended  Status = "Suspended"
)

func NormalizeStatus 

func NormalizeStatus(raw string) Status

NormalizeStatus maps a durabletask ORCHESTRATION_STATUS_* string onto the six dashboard statuses. Unknown / not-yet-started values map to Pending.

func (Status) IsTerminal 

func (s Status) IsTerminal() bool

IsTerminal reports whether a status is final (no further events expected).

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.