executor

type DriveResult ¶

type DriveResult string

DriveResult is a typed enum classifying a single driveOne completion. The values are stable wire-facing strings used as the saga_drive_total{result} metric label. Value set frozen by SAGA-METRIC-LABEL-VALUES-FROZEN-01 (tools/archtest/saga_invariants_test.go).

const (
	// DriveOK means driveOne returned nil (instance advanced cleanly).
	DriveOK DriveResult = "ok"
	// DriveError means driveOne returned a non-nil error (stale lease, instance
	// gone, step failure surfaced to the coordinator, etc.).
	DriveError DriveResult = "error"
)

type Executor ¶

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

Executor drives a single step of a saga.

func NewExecutor(hb Heartbeater, clk clock.Clock, opts ...Option) (*Executor, error)

func (e *Executor) Compensate(
	ctx context.Context,
	inst *ksaga.Instance,
	leaseID idutil.SafeID,
	step ksaga.Step,
	committedState []byte,
) error

func (e *Executor) Execute(
	ctx context.Context,
	inst *ksaga.Instance,
	leaseID idutil.SafeID,
	step ksaga.Step,
	defPolicy ksaga.RetryPolicy,
	prevState []byte,
) Result

func (e *Executor) RunWithHeartbeat(
	ctx context.Context,
	inst *ksaga.Instance,
	leaseID idutil.SafeID,
	fn func(ctx context.Context) error,
) error

type HeartbeatFailureReason ¶

type HeartbeatFailureReason string

HeartbeatFailureReason is a typed enum of why a heartbeat tick reported failure. The values are stable wire-facing strings used as metric labels (saga_heartbeat_failed_total{reason}) and slog fields.

const (
	// HeartbeatFailureInfraError means Heartbeater returned a non-nil err
	// (transient infra fault — e.g., DB timeout, network blip). The heartbeat
	// goroutine logs Warn and continues to the next tick (fail-open transient).
	HeartbeatFailureInfraError HeartbeatFailureReason = "infra_error"
	// HeartbeatFailureStaleLease means Heartbeater returned ok=false: another
	// coordinator has taken over the instance. The heartbeat goroutine logs
	// Warn, invokes the onStale callback (cancels the run context with
	// errLeaseLost), and returns.
	HeartbeatFailureStaleLease HeartbeatFailureReason = "stale_lease"
)

type Heartbeater ¶

type Heartbeater interface {
	// Heartbeat attempts to renew the lease for the given instance.
	// Returns ok=true when the lease was successfully renewed; ok=false when
	// the lease is stale (another coordinator took over). err indicates an
	// infrastructure failure (retry is appropriate).
	Heartbeat(ctx context.Context, instanceID, leaseID idutil.SafeID, leaseDuration time.Duration) (ok bool, err error)
}

Heartbeater extends a saga instance's lease by calling Heartbeat periodically. The interface matches kernel/saga/journal.Heartbeater (and the Heartbeat method on the full journal.Journal) exactly, so any journal implementation satisfies it structurally without this package importing journal.

This is deliberately a SEPARATE declaration from journal.Heartbeater, not an alias or re-use: kernel/ cannot depend on runtime/, and this package must never import kernel/saga/journal (below), so the two structurally-identical interfaces are an intentional layering artifact. journal.Heartbeater exists only to compose the full journal.Journal; this one is the executor's own dependency surface.

INVARIANT: executor never imports kernel/saga/journal — it uses this narrow interface instead. Enforced by the saga-executor-no-journal-import depguard rule in .golangci.yml (path-level import ban) and complemented by SAGA-JOURNAL-HOLDER-SEAL-01 (no journal.* field may be persisted here either).

Caller contract: the Executor stops the heartbeat loop when ok=false is returned; implementations should guarantee that ok=false is idempotent (i.e., repeated observation after the lease is lost is safe).

type LeaderSkipReason ¶

type LeaderSkipReason string

LeaderSkipReason is a typed enum of why the leader-elect gate skipped an instance (acquireLead returned lead=false). The values are stable wire-facing strings used as the saga_leader_elect_skip_total{reason} metric label and as slog fields. Value set frozen by SAGA-METRIC-LABEL-VALUES-FROZEN-01 (tools/archtest/saga_invariants_test.go).

const (
	// LeaderSkipContended means another coordinator holds the per-instance
	// distlock (ErrDistlockTimeout) — expected in multi-process deployments.
	LeaderSkipContended LeaderSkipReason = "contended"
	// LeaderSkipCtxCanceled means the acquire was aborted by ctx cancellation or
	// deadline (normal Stop()/shutdown of the tick loop).
	LeaderSkipCtxCanceled LeaderSkipReason = "ctx_canceled"
	// LeaderSkipBackendError means a distlock backend I/O fault prevented the
	// acquire — this is the lock-acquire failure rate (operationally interesting).
	LeaderSkipBackendError LeaderSkipReason = "backend_error"
)

type NopObserver ¶

type NopObserver struct{}

NopObserver is the zero-cost default Observer. All methods are no-ops.

func (NopObserver) ObserveDrive(_ context.Context, _ string, _ DriveResult)

func (NopObserver) ObserveHeartbeatFailure(_ context.Context, _ idutil.SafeID, _ idutil.SafeID, _ HeartbeatFailureReason)

func (NopObserver) ObserveLeaderSkip(_ context.Context, _ string, _ LeaderSkipReason)

func (NopObserver) ObserveOutcome(_ context.Context, _ idutil.SafeID, _ idutil.SafeID, _, _ string, _ Outcome, _ int)

func (NopObserver) ObserveRetry(_ context.Context, _ idutil.SafeID, _ idutil.SafeID, _, _ string)

func (NopObserver) ObserveTick(_ context.Context, _ TickResult)

type Observer ¶

type Observer interface {
	// ObserveOutcome is called exactly once per Execute call, at the terminal
	// Result. attempts is Result.Attempts (total invocations).
	// instanceID and leaseID identify the specific instance execution for
	// audit / tracing — see Observer godoc for cardinality guidance.
	ObserveOutcome(
		ctx context.Context,
		instanceID, leaseID idutil.SafeID,
		definitionID, stepName string,
		outcome Outcome,
		attempts int,
	)

	// ObserveRetry is called between attempts, before the backoff Sleep starts.
	// Not called for the first attempt (only when attempt N > 1 is about to run).
	// instanceID and leaseID identify the specific instance execution for
	// audit / tracing — see Observer godoc for cardinality guidance.
	ObserveRetry(
		ctx context.Context,
		instanceID, leaseID idutil.SafeID,
		definitionID, stepName string,
	)

	// ObserveHeartbeatFailure is called on each heartbeat tick that fails or
	// observes a stale lease. reason is one of HeartbeatFailureReason constants.
	// instanceID and leaseID identify the specific instance execution for
	// audit / tracing — see Observer godoc for cardinality guidance.
	ObserveHeartbeatFailure(ctx context.Context, instanceID idutil.SafeID, leaseID idutil.SafeID, reason HeartbeatFailureReason)

	// ObserveTick is called once per Coordinator ClaimPending cycle that runs
	// (the stopping-drain early return is not counted). result classifies the
	// cycle: claimed (≥1 instance), empty (0 claimed), or error (ClaimPending
	// failed). Coordinator-emitted — definition_id is not available pre-claim.
	ObserveTick(ctx context.Context, result TickResult)

	// ObserveDrive is called once per driveOne completion with the per-instance
	// outcome: ok (driveOne returned nil) or error (non-nil). Coordinator-emitted;
	// definition_id is the bounded label dimension (per-instance IDs excluded).
	ObserveDrive(ctx context.Context, definitionID string, result DriveResult)

	// ObserveLeaderSkip is called each time the leader-elect gate skips an
	// instance this tick (acquireLead could not confirm leadership). reason is
	// one of LeaderSkipReason constants. Coordinator-emitted; backend_error is
	// the distlock lock-acquire failure rate, contended is normal multi-process
	// contention. definition_id is the bounded label dimension.
	ObserveLeaderSkip(ctx context.Context, definitionID string, reason LeaderSkipReason)
}

Observer is the saga-wide best-effort observability sink (metrics counters, tracing, audit). It carries two families of events:

• Executor-emitted (per-step hot path): ObserveOutcome / ObserveRetry / ObserveHeartbeatFailure — called by the Executor on every Execute outcome / retry attempt / failed heartbeat tick.
• Coordinator-emitted (per-tick / per-drive / per-skip): ObserveTick / ObserveDrive / ObserveLeaderSkip — called by runtime/saga.Coordinator on each ClaimPending cycle, each driveOne, and each leader-elect skip. The contract (interface) lives here for cohesion: the Executor is the lower layer the Coordinator imports, so a single saga observability sink can satisfy both producers (the Coordinator forwards this same value into the internally-constructed Executor via WithObserver).

All methods MUST be non-blocking and must tolerate canceled contexts — they are called on hot paths and must never affect saga correctness.

Concurrency: a single Observer value is shared across every saga instance and is invoked CONCURRENTLY. A Coordinator tick drives its claimed instances in parallel (one goroutine each, bounded by ClaimBatchSize, #983), so both the Coordinator-emitted ObserveDrive and the Executor-emitted per-step methods (ObserveOutcome / ObserveRetry / ObserveHeartbeatFailure) may run for several instances at the same time — the same method on different instances, and different methods, can overlap. Implementations MUST be safe for concurrent use (e.g. atomic counters / mutex-guarded state). The metrics SagaCollector satisfies this: its instruments are concurrency-safe via the OTel SDK.

SHOULD NOT panic; both producers wrap each Observer call in a defer/recover guard so a misbehaving observer logs Warn and execution continues — observability is best-effort and never affects correctness. Recovery is provided by Executor.safeObserveOutcome / safeObserveRetry / safeObserveHeartbeatFailure helpers (executor.go) and Coordinator.safeObserve (coordinator.go).

instanceID and leaseID are carried on the Executor-emitted methods for audit / tracing purposes (e.g. structured log correlation, future audit sink). Metric collectors SHOULD NOT use instanceID or leaseID as label dimensions — they are high-cardinality per-instance values that would cause cardinality explosion in time-series stores. Use definition_id / step_name / outcome / reason / result for metric labels and carry instanceID / leaseID only in log fields or trace span attributes.

ref: temporalio/sdk-go internal_task_handlers.go — server-emitted activity metric envelope (best-effort, non-blocking sink).

type Option ¶

type Option func(*Executor)

Option is a functional option for Executor construction.

func WithHeartbeatInterval(d time.Duration) Option

func WithLeaseDuration(d time.Duration) Option

func WithLogger(l *slog.Logger) Option

func WithObserver(o Observer) Option

func WithObserverCallDeadline(d time.Duration) Option

func WithTracer(t wrapper.Tracer) Option

type Outcome ¶

type Outcome uint8

Outcome classifies the result of an Execute call. The zero value is intentionally invalid — callers must check for known Outcome constants, not treat zero as a sentinel.

const (
	// OutcomeSucceeded means the step's Run returned a nil error.
	OutcomeSucceeded Outcome = iota + 1
	// OutcomeFailed means retry attempts were exhausted with a non-nil error.
	// This is an execution FACT only — the executor does NOT decide whether to
	// compensate. The Coordinator chooses Compensating vs Failed from committed
	// step history (kernel/saga/status.go).
	OutcomeFailed
	// OutcomeExpired means a deadline elapsed: either the per-step Step.Timeout
	// or a parent (saga-level Definition.Timeout) deadline. Terminal.
	OutcomeExpired
	// OutcomeCanceled means the parent context was explicitly canceled
	// (orchestrator shutdown/abort) — NOT a business expiry. The Coordinator
	// should leave the instance for re-claim, not terminate it.
	OutcomeCanceled
	// OutcomeLeaseLost means a heartbeat observed ok=false: another coordinator
	// now owns the lease. The running step was canceled.
	OutcomeLeaseLost
)

func (o Outcome) String() string

type Result ¶

type Result struct {
	// Outcome classifies how Execute ended.
	Outcome Outcome
	// NewState is the opaque JSON state returned by the step's Run function
	// on success (nil on failure/expiry).
	NewState []byte
	// Err is the last error encountered (nil on OutcomeSucceeded).
	Err error
	// Attempts is the total number of Run invocations made.
	Attempts int
}

Result is the value returned by Execute.

type TickResult ¶

type TickResult string

TickResult is a typed enum classifying a Coordinator ClaimPending cycle. The values are stable wire-facing strings used as the saga_tick_total{result} metric label. Value set frozen by SAGA-METRIC-LABEL-VALUES-FROZEN-01 (tools/archtest/saga_invariants_test.go).

const (
	// TickClaimed means ClaimPending returned ≥1 instance this cycle.
	TickClaimed TickResult = "claimed"
	// TickEmpty means ClaimPending returned 0 instances (idle tick).
	TickEmpty TickResult = "empty"
	// TickError means ClaimPending itself failed (journal/backend fault).
	TickError TickResult = "error"
)