tasks/service/history/chasm_task_util.go

package history

import (
	"context"

	enumsspb "github.com/hanzoai/tasks/api/enums/v1"
	"github.com/hanzoai/tasks/chasm"
	"github.com/hanzoai/tasks/service/history/consts"
	historyi "github.com/hanzoai/tasks/service/history/interfaces"
	"github.com/hanzoai/tasks/service/history/tasks"
)

// validateChasmSideEffectTask completes validation of a CHASM side effect task
// after mutable state load/physical task validation.
func validateChasmSideEffectTask(
	ctx context.Context,
	ms historyi.MutableState,
	task *tasks.ChasmTask,
) (any, error) {
	// Because CHASM timers can target closed workflows, we need to specifically
	// exclude zombie workflows, instead of merely checking that the workflow is
	// running.
	if ms.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE {
		return nil, consts.ErrWorkflowZombie
	}

	tree := ms.ChasmTree()
	if tree == nil {
		return nil, errNoChasmTree
	}

	isValid, err := tree.ValidateSideEffectTask(ctx, task)
	if err == nil && isValid {
		// If the task is still valid, keep it around by returning a non-nil value.
		return &struct{}{}, nil
	}

	return nil, err
}

// executeChasmSideEffectTask completes execution of a CHASM side effect task
// after physical task validation.
//
// TODO - ExecuteSideEffectTask doesn't need to be on ChasmTree once path
// decoding is done at access time.
func executeChasmSideEffectTask(
	ctx context.Context,
	engine chasm.Engine,
	registry *chasm.Registry,
	tree historyi.ChasmTree,
	task *tasks.ChasmTask,
) error {
	executionKey := chasm.ExecutionKey{
		NamespaceID: task.NamespaceID,
		BusinessID:  task.WorkflowID,
		RunID:       task.RunID,
	}

	validate := func(backend chasm.NodeBackend, _ chasm.Context, _ chasm.Component) error {
		// Because CHASM timers can target closed workflows, we need to specifically
		// exclude zombie workflows, instead of merely checking that the workflow is
		// running.
		if backend.GetExecutionState().State == enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE {
			return consts.ErrWorkflowZombie
		}

		// Validate task generation. We don't need to refresh tasks as we re-generate
		// CHASM tasks on transaction close.
		taskID := task.TaskID
		tgClock := backend.GetExecutionInfo().TaskGenerationShardClockTimestamp
		if tgClock != 0 && taskID != 0 && taskID < tgClock {
			return consts.ErrStaleReference
		}

		return nil
	}

	engineCtx := chasm.NewEngineContext(ctx, engine)
	return tree.ExecuteSideEffectTask(
		engineCtx,
		registry,
		executionKey,
		task,
		validate,
	)
}