tasks/service/history/hsm/tasks.go

package hsm

import (
	"fmt"
	"time"

	persistencespb "go.temporal.io/server/api/persistence/v1"
	"go.temporal.io/server/service/history/consts"
)

// Immediate is a sentinel value to express that a task must be executed immediately instead of delayed to a certain
// deadline.
var Immediate = time.Time{}

// A Task is generated by a state machine in order to drive execution. For example, a callback state machine in the
// SCHEDULED state, would generate an invocation task to be eventually executed by the framework. State machine
// transitions and tasks are committed atomically to ensure that the system is in a consistent state.
//
// Tasks are generated by calling the [StateMachine.Tasks] method on a state machine after it has transitioned. Tasks
// are executed by an executor that is registered to handle a specific task type. The framework converts this minimal
// task representation into [tasks.Task] instances, filling in the state machine reference, workflow key, and task ID.
// A [TaskSerializer] need to be registered in a [Registry] for a given type in order to process tasks of that type.
type Task interface {
	// Task type that must be unique per task definition.
	Type() string
	// A deadline for firing this task.
	// This represents a lower bound and actual execution may get delayed if the system is overloaded or for various
	// other reasons.
	// Return [Immediate] to schedule this task immediately.
	Deadline() time.Time
	// The destination of this task, used to group tasks into a per namespace-and-destination scheduler.
	// Tasks may return an empty string if the task is delivered internally within the system.
	// If a destination is set, a task will be scheduled on the outbound queue.
	// Currently Destination and Deadline are mutually exclusive.
	Destination() string
	// Validate checks if the task is still valid for processing for the current node state.
	// Implementors may return [ErrStaleReference] or [consts.ErrWorkflowCompleted] if the task is no longer valid.
	// A typical implementation may use [node.CheckRunning], [ValidateNotTransitioned], or check if the state of the
	// machine is still relevant for running this task.
	Validate(ref *persistencespb.StateMachineRef, node *Node) error
}

// TaskAttributes are common attributes for a [Task].
type TaskAttributes struct {
	Deadline    time.Time
	Destination string
}

// TaskSerializer provides type information and a serializer for a state machine.
type TaskSerializer interface {
	Serialize(Task) ([]byte, error)
	Deserialize(data []byte, attrs TaskAttributes) (Task, error)
}

// ValidateNotTransitioned returns a [consts.ErrStaleReference] if the machine has transitioned since the task was
// generated.
func ValidateNotTransitioned(ref *persistencespb.StateMachineRef, node *Node) error {
	if ref.MachineTransitionCount != node.InternalRepr().TransitionCount {
		return fmt.Errorf("%w: state machine transitions (%d) != ref transitions (%d)", consts.ErrStaleReference, node.InternalRepr().TransitionCount, ref.MachineTransitionCount)
	}
	return nil
}

// ValidateState returns a [consts.ErrStaleReference] if the machine is not in the expected state.
func ValidateState[S comparable, T StateMachine[S]](node *Node, expected S) error {
	cb, err := MachineData[T](node)
	if err != nil {
		return err
	}
	if cb.State() != expected {
		return fmt.Errorf(
			"%w: %w: expected a %s machine in %v state, got %v",
			consts.ErrStaleReference,
			ErrInvalidTransition,
			node.Key.ID,
			expected,
			cb.State(),
		)
	}
	return nil
}