executor

Documentation

Overview

Package executor is backplane's side of hook calls (design §7.2): it answers every hook of the installation over Temporal Nexus and runs bindings — and the rules engine's rules — as the workflow wire.BindingWorkflow ("backplane.Binding.v1") on backplane's own task queue.

• Endpoints. For every service whose manifests declare hooks there is a Nexus endpoint named after the service, targeting backplane's queue; the executor creates it as soon as the registry shows such a manifest (not when a binding is saved), so a call without a binding gets "no binding", not "endpoint not found". Creation is idempotent across replicas (AlreadyExists is success); an endpoint pointing elsewhere is updated. Endpoints belonging to this queue are removed after a service has no instances for the absence grace period, with a fresh Consul check. Unhealthy instances still count as present; manifests are retained.

• Nexus handler. Temporal's Go SDK registers Nexus services on a worker before it starts and dispatches by exact service and operation name — there is no catch-all handler — while hook services come and go with the registry. So the handler is a worker of its own: Nexus only (no workflow or activity pollers), on backplane's queue, serving <service>.Hooks with one operation per hook for every hook any known manifest declares. When that set changes the executor starts a new worker with the new set and then stops the old one; backplane's main worker (workflows and activities) is untouched, so its registrations stay static and identical on every replica. The set is derived from the registry alone — every manifest of every service, all versions — so replicas converge on the same one; a new service's endpoint is created only after the local worker serves it and a short settle delay, which covers the other replicas seeing the same registry change.

• Operation. <Hook> of <service>.Hooks takes HookCall and gives HookResult. The binding in force is compiled against the latest manifests and started as wire.BindingWorkflow, id binding/<hook>/<Nexus request id> (a retried start is the same run), execution timeout = HookCall.deadline; the operation is the run (async). No binding: a required hook fails with a non-retryable application error of type backplane.NoBinding; an optional one answers {} at once.

• Console. TestBinding, ListBindingRuns, GetBindingRun and CancelBindingRun of BindingService (BindingService wraps the bindings Manager's API with them).

Index

• Constants
• Variables
• func RegisterWorkflow(r worker.Registry)
• type Bindings
• type EventMeta
• type Executor
  • func New(parent deps.Scope, b Bindings, src registry.Source, opts ...Option) *Executor
  • func (x *Executor) BindingService(base consolev1.BindingServiceServer) consolev1.BindingServiceServer
  • func (x *Executor) Queue() string
  • func (x *Executor) Register(base consolev1.BindingServiceServer) func(r grpc.ServiceRegistrar)
  • func (x *Executor) RegisterWorkflows(r worker.Registry)
  • func (x *Executor) RunBinding(ctx context.Context, hook string, prog *bindings.Program, version int64, ...) (*backplanev1.HookResult, error)
  • func (x *Executor) Sync(ctx context.Context) error
  • func (x *Executor) Temporal() *direct.TemporalBindings
• type Identity
  • func Timeline(ctx context.Context, c client.Client, run *consolev1.GetRunResponse) ([]*consolev1.StepRun, Identity, error)
  • func (i Identity) String() string
• type Input
  • func BindingInput(p *bindings.Program, id Identity, payload []byte, trace map[string]string) (Input, error)
  • func RuleInput(p *bindings.Program, id Identity, payload []byte, meta EventMeta, ...) (Input, error)
• type Kind
• type Option
  • func AbsenceGrace(d time.Duration) Option
  • func Author(fn func(ctx context.Context) string) Option
  • func Namespace(ns string) Option
  • func Resync(d time.Duration) Option
  • func Runs(runs consolev1.WorkflowServiceServer) Option
  • func Settle(d time.Duration) Option
  • func WithDirect(d *direct.Service) Option
  • func WithTemporal(fn TemporalFunc, queue string) Option
• type TemporalFunc

Constants

const (
	// DefaultSettle is how long a new hook service is served before its
	// endpoint is created.
	DefaultSettle = 2 * time.Second
	// DefaultResync is how often endpoints are checked without a registry
	// change (one deleted by hand comes back).
	DefaultResync = time.Minute
	// DefaultAbsenceGrace protects brief restarts and discovery propagation.
	DefaultAbsenceGrace = 5 * time.Minute
)

Defaults.

Variables

var (
	// ErrNotSynced: the registry has no snapshot yet.
	ErrNotSynced = errors.New("executor: registry not synced yet")
	// ErrUnavailable: Temporal is not configured or not connected.
	ErrUnavailable = errors.New("executor: temporal unavailable")
)

Errors.

Functions

func RegisterWorkflow

func RegisterWorkflow(r worker.Registry)

RegisterWorkflow adds wire.BindingWorkflow to r without metrics: tests and tools. backplane's worker takes Executor.RegisterWorkflows.

Types

type Bindings

type Bindings interface {
	// Active is the binding in force and its version; ErrNoBinding.
	Active(ctx context.Context, hook string) (bindings.Binding, int64, error)
	// Version is one version, 0 the current one.
	Version(ctx context.Context, hook string, version int64) (bindings.BindingVersion, error)
}

Bindings is what the executor reads of bindings: *bindings.Manager.

type EventMeta

type EventMeta struct {
	ID      string    `json:"id,omitempty"`
	Source  string    `json:"source,omitempty"`
	Subject string    `json:"subject,omitempty"`
	Type    string    `json:"type,omitempty"`
	Time    time.Time `json:"time"`
}

EventMeta are the CloudEvents attributes of a rule's event.

type Executor

type Executor struct {
	deps.Component
	// contains filtered or unexported fields
}

Executor is the component: the Nexus worker, the endpoints, the compile cache. It holds state and goroutines: share it by pointer.

func New

func New(parent deps.Scope, b Bindings, src registry.Source, opts ...Option) *Executor

New creates the component under parent: bindings from b, manifests from src, Temporal through the SDK's client. Give RegisterWorkflows to workflows.Register on the same root.

func (*Executor) BindingService

func (x *Executor) BindingService(base consolev1.BindingServiceServer) consolev1.BindingServiceServer

BindingService is base (bindings.Manager's BindingAPI) with the run RPCs: the direct runtime's, with the executor as their Temporal side.

func (*Executor) Queue

func (x *Executor) Queue() string

Queue is backplane's task queue: the endpoints' target, where runs go.

func (*Executor) Register

func (x *Executor) Register(base consolev1.BindingServiceServer) func(r grpc.ServiceRegistrar)

Register registers BindingService — base with the run RPCs — on r.

func (*Executor) RegisterWorkflows

func (x *Executor) RegisterWorkflows(r worker.Registry)

RegisterWorkflows adds wire.BindingWorkflow to backplane's worker (give it to workflows.Register).

func (*Executor) RunBinding

func (x *Executor) RunBinding(
	ctx context.Context, hook string, prog *bindings.Program, version int64, call *backplanev1.HookCall,
) (*backplanev1.HookResult, error)

RunBinding runs a hook call's Temporal-runtime binding and waits for its answer: the direct runtime's TemporalRunner, for calls that arrive over HookService.

func (*Executor) Sync

func (x *Executor) Sync(ctx context.Context) error

Sync makes the Nexus worker serve the hooks of the current registry snapshot and ensures their endpoints. follow runs it; tests call it.

func (*Executor) Temporal

func (x *Executor) Temporal() *direct.TemporalBindings

Temporal is the executor as the Temporal side of the run RPCs.

type Identity

type Identity struct {
	// Hook is the full hook name of a binding.
	Hook string `json:"hook,omitempty"`
	// Rule is the id of a rule.
	Rule string `json:"rule,omitempty"`
	// Version of the binding or rule; 0: an unsaved definition (a test).
	Version int64 `json:"version,omitempty"`
	// Test: a console test run.
	Test bool `json:"test,omitempty"`
}

Identity is what a run executes.

func Timeline

func Timeline(
	ctx context.Context, c client.Client, run *consolev1.GetRunResponse,
) ([]*consolev1.StepRun, Identity, error)

Timeline is the steps of a run of wire.BindingWorkflow — a binding's or a rule's — and what ran: every step of the program in its input by name, its call from the run's history (c reads it) or NOT_RUN, then the compensations in the order they were scheduled.

func (Identity) String

func (i Identity) String() string

String is "<hook>@<version>" or "rule:<id>@<version>"; an unsaved definition is "@draft": ActivityCall.binding and the memo.

type Input

type Input struct {
	Kind Kind `json:"kind"`
	// Program is a marshaled *bindings.Program (CompileBinding or
	// CompileRule) of the same kind.
	Program json.RawMessage `json:"program"`
	// Identity names what runs: ActivityCall.binding, the memo, metrics.
	Identity Identity `json:"identity"`
	// Payload is the hook's input (`req`) or the event's payload
	// (`event`): JSON; empty is {}.
	Payload json.RawMessage `json:"payload,omitempty"`
	// Meta are the event's CloudEvents attributes (`meta`); rules only.
	Meta *EventMeta `json:"meta,omitempty"`
	// Trace is the caller's W3C trace context (HookCall.trace, the event's
	// headers): ActivityCall.trace of every step when the workflow has no
	// span of its own.
	Trace map[string]string `json:"trace,omitempty"`
}

Input is the input of the workflow wire.BindingWorkflow: everything a run needs, nothing it has to look up — the compiled program travels with it, so a run is not affected by saves or manifests that come after its start, and replays evaluate exactly what was compiled.

The rules engine starts the same workflow with Kind KindRule, the compiled rule, Meta and Identity.Rule.

func BindingInput

func BindingInput(p *bindings.Program, id Identity, payload []byte, trace map[string]string) (Input, error)

BindingInput is the input of a binding's run: the program compiled from the binding, the hook's input and the caller's trace.

func RuleInput

func RuleInput(
	p *bindings.Program, id Identity, payload []byte, meta EventMeta, trace map[string]string,
) (Input, error)

RuleInput is the input of a rule's run on an event.

type Kind

type Kind string

Kind is what a run executes.

const (
	// KindBinding: a hook call; the input is `req`, the result the hook's
	// output.
	KindBinding Kind = "binding"
	// KindRule: an event; the input is `event` with `meta`, no result.
	KindRule Kind = "rule"
)

Kinds of runs.

type Option

type Option func(*Executor)

Option configures New.

func AbsenceGrace

func AbsenceGrace(d time.Duration) Option

AbsenceGrace sets how long a service must have no instances before its Nexus endpoint is removed. Zero is useful in tests; negative values panic.

func Author

func Author(fn func(ctx context.Context) string) Option

Author names who made a console call from its context; default: the console session as console:<id>, else "admin".

func Namespace

func Namespace(ns string) Option

Namespace of Temporal the endpoints target (default "default"): the SDK's BACKPLANE_TEMPORAL_NS.

func Resync

func Resync(d time.Duration) Option

Resync replaces DefaultResync.

func Runs

func Runs(runs consolev1.WorkflowServiceServer) Option

Runs is how runs are listed, read and canceled: ops' WorkflowService (visibility, history summaries). Without it those RPCs are UNAVAILABLE.

func Settle

func Settle(d time.Duration) Option

Settle replaces DefaultSettle (tests: 0).

func WithDirect

func WithDirect(d *direct.Service) Option

WithDirect installs the direct runtime: direct-runtime bindings called over Nexus run on it, and the run RPCs list its runs first.

func WithTemporal

func WithTemporal(fn TemporalFunc, queue string) Option

WithTemporal replaces the SDK's Temporal client and backplane's task queue (tests).

type TemporalFunc

type TemporalFunc func() (client.Client, error)

TemporalFunc gives the Temporal client.