tasks/service/worker/scanner/executions/task.go

package executions

import (
	"context"
	"time"

	commonpb "go.temporal.io/api/common/v1"
	"go.temporal.io/api/serviceerror"
	"github.com/hanzoai/tasks/api/adminservice/v1"
	"github.com/hanzoai/tasks/api/historyservice/v1"
	persistencespb "github.com/hanzoai/tasks/api/persistence/v1"
	"github.com/hanzoai/tasks/chasm"
	"github.com/hanzoai/tasks/common/backoff"
	"github.com/hanzoai/tasks/common/collection"
	"github.com/hanzoai/tasks/common/dynamicconfig"
	"github.com/hanzoai/tasks/common/log"
	"github.com/hanzoai/tasks/common/log/tag"
	"github.com/hanzoai/tasks/common/metrics"
	"github.com/hanzoai/tasks/common/namespace"
	"github.com/hanzoai/tasks/common/persistence"
	"github.com/hanzoai/tasks/common/quotas"
	"github.com/hanzoai/tasks/service/worker/scanner/executor"
)

const (
	executionsPageSize = 100

	taskStartupDelayRatio              = 100 * time.Millisecond
	taskStartupDelayRandomizationRatio = 1.0
)

type (
	// task is a runnable task that adheres to the executor.Task interface
	// for the scavenger, each of this task processes a single workflow mutableState
	task struct {
		shardID          int32
		executionManager persistence.ExecutionManager
		registry         namespace.Registry
		historyClient    historyservice.HistoryServiceClient
		adminClient      adminservice.AdminServiceClient
		metricsHandler   metrics.Handler
		logger           log.Logger
		scavenger        *Scavenger

		ctx                           context.Context
		rateLimiter                   quotas.RateLimiter
		executionDataDurationBuffer   dynamicconfig.DurationPropertyFn
		enableHistoryEventIDValidator dynamicconfig.BoolPropertyFn
		paginationToken               []byte
	}
)

// newTask returns a new instance of an executable task which will process a single mutableState
func newTask(
	ctx context.Context,
	shardID int32,
	executionManager persistence.ExecutionManager,
	registry namespace.Registry,
	historyClient historyservice.HistoryServiceClient,
	adminClient adminservice.AdminServiceClient,
	metricsHandler metrics.Handler,
	logger log.Logger,
	scavenger *Scavenger,
	rateLimiter quotas.RateLimiter,
	executionDataDurationBuffer dynamicconfig.DurationPropertyFn,
	enableHistoryEventIDValidator dynamicconfig.BoolPropertyFn,
) executor.Task {
	return &task{
		shardID:          shardID,
		executionManager: executionManager,
		registry:         registry,
		historyClient:    historyClient,
		adminClient:      adminClient,

		metricsHandler: metricsHandler.WithTags(metrics.OperationTag(metrics.ExecutionsScavengerScope)),
		logger:         logger,
		scavenger:      scavenger,

		ctx:                           ctx,
		rateLimiter:                   rateLimiter,
		executionDataDurationBuffer:   executionDataDurationBuffer,
		enableHistoryEventIDValidator: enableHistoryEventIDValidator,
	}
}

// Run runs the task
func (t *task) Run() executor.TaskStatus {
	time.Sleep(backoff.Jitter(
		taskStartupDelayRatio*time.Duration(t.scavenger.numHistoryShards),
		taskStartupDelayRandomizationRatio,
	))

	iter := collection.NewPagingIteratorWithToken(t.getPaginationFn(), t.paginationToken)
	var retryTask bool
	for iter.HasNext() {
		_ = t.rateLimiter.Wait(t.ctx)
		record, err := iter.Next()
		if err != nil {
			metrics.ScavengerValidationSkipsCount.With(t.metricsHandler).Record(1)
			// break out of the loop when pagination fails
			t.logger.Error("unable to paginate concrete execution", tag.ShardID(t.shardID), tag.Error(err))
			retryTask = true
			break
		}

		mutableState := &MutableState{WorkflowMutableState: record}
		results := t.validate(mutableState)
		printValidationResult(
			mutableState,
			results,
			t.metricsHandler,
			t.logger,
		)
		err = t.handleFailures(mutableState, results)
		if err != nil {
			// continue validation process and retry after all workflow records has been iterated.
			executionInfo := mutableState.GetExecutionInfo()
			metrics.ScavengerValidationSkipsCount.With(t.metricsHandler).Record(1)
			t.logger.Error("unable to process failure result",
				tag.ShardID(t.shardID),
				tag.Error(err),
				tag.WorkflowNamespaceID(executionInfo.GetNamespaceId()),
				tag.WorkflowID(executionInfo.GetWorkflowId()),
				tag.WorkflowRunID(mutableState.GetExecutionState().GetRunId()))
			retryTask = true
		}
	}
	if retryTask {
		return executor.TaskStatusDefer
	}
	return executor.TaskStatusDone
}

func (t *task) validate(
	mutableState *MutableState,
) []MutableStateValidationResult {

	var results []MutableStateValidationResult
	t.logger.Debug("validating mutable state",
		tag.ShardID(t.shardID),
		tag.WorkflowNamespaceID(mutableState.GetExecutionInfo().GetNamespaceId()),
		tag.WorkflowID(mutableState.GetExecutionInfo().GetWorkflowId()),
		tag.WorkflowRunID(mutableState.GetExecutionState().GetRunId()),
	)

	if validationResults, err := NewMutableStateValidator(
		t.registry,
		t.executionDataDurationBuffer,
	).Validate(
		t.ctx,
		mutableState,
	); err != nil {
		t.logger.Error("unable to validate mutable state ID",
			tag.ShardID(t.shardID),
			tag.WorkflowNamespaceID(mutableState.GetExecutionInfo().GetNamespaceId()),
			tag.WorkflowID(mutableState.GetExecutionInfo().GetWorkflowId()),
			tag.WorkflowRunID(mutableState.GetExecutionState().GetRunId()),
			tag.Error(err),
		)
	} else {
		results = append(results, validationResults...)
	}

	// Fail fast if the mutable is corrupted, no need to validate history.
	if len(results) > 0 {
		return results
	}

	if t.enableHistoryEventIDValidator() {
		if validationResults, err := NewHistoryEventIDValidator(
			t.shardID,
			t.executionManager,
		).Validate(t.ctx, mutableState); err != nil {
			t.logger.Error("unable to validate history event ID being contiguous",
				tag.ShardID(t.shardID),
				tag.WorkflowNamespaceID(mutableState.GetExecutionInfo().GetNamespaceId()),
				tag.WorkflowID(mutableState.GetExecutionInfo().GetWorkflowId()),
				tag.WorkflowRunID(mutableState.GetExecutionState().GetRunId()),
				tag.Error(err),
			)
		} else {
			results = append(results, validationResults...)
		}
	}

	return results
}

func (t *task) getPaginationFn() collection.PaginationFn[*persistencespb.WorkflowMutableState] {
	return func(paginationToken []byte) ([]*persistencespb.WorkflowMutableState, []byte, error) {
		req := &persistence.ListConcreteExecutionsRequest{
			ShardID:   t.shardID,
			PageSize:  executionsPageSize,
			PageToken: paginationToken,
		}
		resp, err := t.executionManager.ListConcreteExecutions(t.ctx, req)
		if err != nil {
			return nil, nil, err
		}
		paginateItems := resp.States
		t.paginationToken = resp.PageToken
		return paginateItems, resp.PageToken, nil
	}
}

func (t *task) handleFailures(
	mutableState *MutableState,
	results []MutableStateValidationResult,
) error {
	for _, failure := range results {
		switch failure.failureType {
		case mutableStateRetentionFailureType:
			executionInfo := mutableState.GetExecutionInfo()
			runID := mutableState.GetExecutionState().GetRunId()
			ns, err := t.registry.GetNamespaceByID(namespace.ID(executionInfo.GetNamespaceId()))
			switch err.(type) {
			case *serviceerror.NotFound,
				*serviceerror.NamespaceNotFound:
				t.logger.Error("Garbage data in DB after namespace is deleted", tag.WorkflowNamespaceID(executionInfo.GetNamespaceId()))
				// We cannot do much in this case. It just ignores this error.
				return nil
			case nil:
				// continue to delete
			default:
				return err
			}

			// TODO: consider using history force delete api directly and extract correct archetypeID from mutableState.
			// Currently, checks on chasm executions are bypassed and we don't reach here for chasm executions.
			_, err = t.adminClient.DeleteWorkflowExecution(t.ctx, &adminservice.DeleteWorkflowExecutionRequest{
				Namespace: ns.Name().String(),
				Execution: &commonpb.WorkflowExecution{
					WorkflowId: executionInfo.GetWorkflowId(),
					RunId:      runID,
				},
				Archetype: chasm.WorkflowArchetype,
			})
			switch err.(type) {
			case *serviceerror.NotFound,
				*serviceerror.NamespaceNotFound:
				return nil
			case nil:
				continue
			default:
				return err
			}
		default:
			// no-op
			continue
		}
	}
	return nil
}

func printValidationResult(
	mutableState *MutableState,
	results []MutableStateValidationResult,
	metricsHandler metrics.Handler,
	logger log.Logger,
) {
	metrics.ScavengerValidationRequestsCount.With(metricsHandler).Record(1)
	if len(results) == 0 {
		return
	}

	metrics.ScavengerValidationFailuresCount.With(metricsHandler).Record(1)
	for _, result := range results {
		metrics.ScavengerValidationFailuresCount.With(metricsHandler).Record(1, metrics.FailureTag(result.failureType))
		logger.Info(
			"validation failed for execution.",
			tag.WorkflowNamespaceID(mutableState.GetExecutionInfo().GetNamespaceId()),
			tag.WorkflowID(mutableState.GetExecutionInfo().GetWorkflowId()),
			tag.WorkflowRunID(mutableState.GetExecutionState().GetRunId()),
			tag.Value(result.failureDetails),
		)
	}
}