runtime

Documentation

Index

• type LocalOption
  • func WithLocalTool(t tool.Tool) LocalOption
• type Option
  • func WithAwaitBindingRepo(r repository.AwaitBindingRepository) Option
  • func WithCostRecorder(r cost.Recorder) Option
  • func WithDB(db *gorm.DB) Option
  • func WithEventBus(b *eventbus.EventBus) Option
  • func WithJobQueue(q engine.AsyncJobQueue) Option
  • func WithLock(l lock.DistributedLock) Option
  • func WithQueue(q repository.TaskQueue) Option
  • func WithTool(t tool.Tool) Option
• type Options
• type RunResult
• type Runtime
  • func New(opts ...Option) (*Runtime, error)
  • func NewLocal(sqlitePath string, opts ...LocalOption) (*Runtime, error)
  • func (r *Runtime) DB() *gorm.DB
  • func (r *Runtime) Engine() *engine.Engine
  • func (r *Runtime) EventBus() *eventbus.EventBus
  • func (r *Runtime) NodeRegistry() *nodes.NodeRegistry
  • func (r *Runtime) RegisterWorkflow(ctx context.Context, def *definition.WorkflowDefinition) error
  • func (r *Runtime) Resume(ctx context.Context, taskID int64, nodeName string, meta map[string]any) (*RunResult, error)
  • func (r *Runtime) Retry(ctx context.Context, taskID int64, resumeFrom string, ...) error
  • func (r *Runtime) Run(ctx context.Context, def *definition.WorkflowDefinition, input map[string]any) (*RunResult, error)
  • func (r *Runtime) Shutdown() error
  • func (r *Runtime) Start(ctx context.Context, opts ...StartOption) error
  • func (r *Runtime) Status(ctx context.Context, taskID int64) (*domain.Task, error)
  • func (r *Runtime) Submit(ctx context.Context, workflowName string, input map[string]any) (int64, error)
  • func (r *Runtime) Subscribe(eventType string) (<-chan *domain.TaskEvent, func())
  • func (r *Runtime) ToolRegistry() *tool.Registry
  • func (r *Runtime) WorkflowRegistry() *registry.WorkflowRegistry
• type StartOption
  • func WithAsyncWorkers(n int) StartOption
  • func WithAwaitPollWorkers(n int) StartOption
  • func WithTaskWorkers(n int) StartOption
• type StartOptions

Types

type LocalOption

type LocalOption func(*localOptions)

LocalOption configures a local (single-process) Runtime.

func WithLocalTool

func WithLocalTool(t tool.Tool) LocalOption

WithLocalTool registers a tool for local mode.

type Option

type Option func(*Options)

Option configures a server-side Runtime.

func WithAwaitBindingRepo

func WithAwaitBindingRepo(r repository.AwaitBindingRepository) Option

WithAwaitBindingRepo sets the await binding repository (enables await nodes).

func WithCostRecorder

func WithCostRecorder(r cost.Recorder) Option

WithCostRecorder sets a custom cost recorder.

func WithDB

func WithDB(db *gorm.DB) Option

WithDB sets the GORM database handle.

func WithEventBus

func WithEventBus(b *eventbus.EventBus) Option

WithEventBus sets the event bus.

func WithJobQueue

func WithJobQueue(q engine.AsyncJobQueue) Option

WithJobQueue sets the async job queue.

func WithLock

func WithLock(l lock.DistributedLock) Option

WithLock sets the distributed lock.

func WithQueue

func WithQueue(q repository.TaskQueue) Option

WithQueue sets the task queue.

func WithTool

func WithTool(t tool.Tool) Option

WithTool registers a tool for use in tool-type workflow nodes.

type Options

type Options struct {
	DB    *gorm.DB
	Queue repository.TaskQueue
	Lock  lock.DistributedLock
	Bus   *eventbus.EventBus
	JobQ  engine.AsyncJobQueue
	// contains filtered or unexported fields
}

Options holds all externally-provided dependencies.

type RunResult

type RunResult struct {
	TaskID        int64
	Status        string
	Err           error
	Task          *domain.Task
	SuspendReason string
	SuspendNode   string
}

RunResult represents the outcome of a Runtime.Run call.

type Runtime

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

Runtime is the one-stop entry point for the flux-workflow engine.

Two constructors cover all use cases:

// Local mode — self-contained, SQLite, no external dependencies
r, _ := runtime.NewLocal("./state.db")

// Server mode — external DB, queue, lock, event bus
r, _ := runtime.New(
    runtime.WithDB(pgDB),
    runtime.WithQueue(redisQ),
    runtime.WithEventBus(bus),
    runtime.WithJobQueue(jobQ),
)

After creation, Run executes a workflow definition synchronously, or Start launches background workers so Submit-ed tasks execute asynchronously.

func New

func New(opts ...Option) (*Runtime, error)

New creates a Runtime from externally-provided dependencies.

func NewLocal

func NewLocal(sqlitePath string, opts ...LocalOption) (*Runtime, error)

NewLocal creates a self-contained Runtime using SQLite.

All dependencies are created internally:

• SQLite database (WAL mode, auto-migrate)
• In-memory task queue
• In-memory async job queue
• In-memory distributed lock
• Event bus (no-op persistence and WS, for local use)

Tools can be registered via WithLocalTool.

func (*Runtime) DB

func (r *Runtime) DB() *gorm.DB

DB exposes the underlying *gorm.DB. With it a consumer can construct any query repository (via repository/query) or run its own read-side queries — the primary extension point for building a business/HTTP layer on top.

func (*Runtime) Engine

func (r *Runtime) Engine() *engine.Engine

Engine exposes the underlying engine for advanced use (replay, redo, cancel, fork, and other power-user operations that are intentionally not on the facade).

func (*Runtime) EventBus

func (r *Runtime) EventBus() *eventbus.EventBus

EventBus exposes the event bus, for publishing custom events or attaching additional listeners beyond Subscribe.

func (*Runtime) NodeRegistry

func (r *Runtime) NodeRegistry() *nodes.NodeRegistry

NodeRegistry exposes the node registry for registering custom node types.

func (*Runtime) RegisterWorkflow

func (r *Runtime) RegisterWorkflow(ctx context.Context, def *definition.WorkflowDefinition) error

RegisterWorkflow 注册工作流定义并立即持久化。 定义变更时自动发布新版本（按 def.Hash() 判定）；此后可通过 Submit(ctx, def.Name, input) 按名称提交任务。

func (*Runtime) Resume

func (r *Runtime) Resume(
	ctx context.Context,
	taskID int64,
	nodeName string,
	meta map[string]any,
) (*RunResult, error)

Resume 用外部事件的结果唤醒挂起中的任务：把 meta 作为 nodeName 节点的输出 闭合该节点，然后在当前调用内同步续跑 DAG，直到下一个终态或挂起点。

典型场景：await/async 节点挂起后，外部回调（webhook、人工审批、异步任务完成） 携带结果到达。节点已被处理时返回 Status "noop"，可安全重复调用。 failed/canceled 的任务请使用 Retry。

func (*Runtime) Retry

func (r *Runtime) Retry(
	ctx context.Context,
	taskID int64,
	resumeFrom string,
	patches []domain.RuntimePatch,
) error

Retry 人工恢复 failed / canceled / suspended 的任务：重置失败子树后重新入队， 由后台 worker（需已 Start）异步执行。

resumeFrom 指定从某节点重跑（空则自动收集失败根节点）；patches 可在恢复前 修正上游节点的输出。手动重试会重置自动重试计数器。

func (*Runtime) Run

func (r *Runtime) Run(
	ctx context.Context,
	def *definition.WorkflowDefinition,
	input map[string]any,
) (*RunResult, error)

Run executes a workflow definition synchronously.

The definition does not need to be registered. If it is registered and unchanged (same hash), the task is linked to the persisted version so it can later be resumed/replayed by version ID.

func (*Runtime) Shutdown

func (r *Runtime) Shutdown() error

Shutdown stops background workers (if started), waits for them to exit, then closes the database connection if owned by this Runtime.

func (*Runtime) Start

func (r *Runtime) Start(ctx context.Context, opts ...StartOption) error

Start launches the background workers that consume Submit-ed tasks:

• task workers: 从任务队列取任务并交给 engine 执行
• recovery scanner: 扫描 crash 的 running 节点并自动恢复
• async workers: 消费异步节点 job 队列
• await poll workers: 轮询 await binding 的超时与到期

Workers run until Shutdown is called or ctx is canceled. Start 只能调用一次；未调用 Start 时 Run 仍可同步执行工作流。

func (*Runtime) Status

func (r *Runtime) Status(ctx context.Context, taskID int64) (*domain.Task, error)

Status returns the current task state.

func (*Runtime) Submit

func (r *Runtime) Submit(
	ctx context.Context,
	workflowName string,
	input map[string]any,
) (int64, error)

Submit enqueues a task for async execution by workflow name and returns the task ID. The workflow must have been registered via RegisterWorkflow (or otherwise persisted); the task is bound to the latest version, which workers resolve to load the definition.

func (*Runtime) Subscribe

func (r *Runtime) Subscribe(eventType string) (<-chan *domain.TaskEvent, func())

Subscribe returns a channel that receives events for the given event type, plus a cancel function that unsubscribes and closes the channel.

func (*Runtime) ToolRegistry

func (r *Runtime) ToolRegistry() *tool.Registry

ToolRegistry exposes the tool registry the engine's tool nodes resolve against. Register additional tools on it (they resolve lazily at node execution) and share it with any service that executes tools directly, so the engine and the business layer see the same tool set.

func (*Runtime) WorkflowRegistry

func (r *Runtime) WorkflowRegistry() *registry.WorkflowRegistry

type StartOption

type StartOption func(*StartOptions)

StartOption configures Runtime.Start.

func WithAsyncWorkers

func WithAsyncWorkers(n int) StartOption

WithAsyncWorkers sets the number of async node job workers.

func WithAwaitPollWorkers

func WithAwaitPollWorkers(n int) StartOption

WithAwaitPollWorkers sets the number of await poll workers.

func WithTaskWorkers

func WithTaskWorkers(n int) StartOption

WithTaskWorkers sets the number of task execution workers.

type StartOptions

type StartOptions struct {
	TaskWorkers      int // 任务执行 worker 数，默认 2
	AsyncWorkers     int // 异步节点 job worker 数，默认 2
	AwaitPollWorkers int // await 轮询 worker 数，默认 1
}

StartOptions 配置后台 worker 并发度。