tasks/service/history/hsm/tasks.go

// The MIT License
//
// Copyright (c) 2024 Temporal Technologies Inc.  All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

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
}