iop/packages/go/agenttask/integration_queue.go

package agenttask

import (
	"context"
	"fmt"
	"reflect"
	"sort"
)

type integrationCandidate struct {
	ProjectID  ProjectID
	WorkUnitID WorkUnitID
	Workspace  WorkspaceID
	Ordinal    DispatchOrdinal
}

func (m *Manager) integratePending(
	ctx context.Context,
	active []ProjectID,
) (bool, error) {
	state, err := m.load(ctx)
	if err != nil {
		return false, err
	}
	var candidates []integrationCandidate
	for _, projectID := range active {
		project := state.Projects[projectID]
		for workID, work := range project.Works {
			if work.State == WorkStatePendingIntegration {
				candidates = append(candidates, integrationCandidate{
					ProjectID: projectID, WorkUnitID: workID,
					Workspace: project.WorkspaceID, Ordinal: work.DispatchOrdinal,
				})
			}
		}
	}
	sort.Slice(candidates, func(left, right int) bool {
		if candidates[left].Ordinal != candidates[right].Ordinal {
			return candidates[left].Ordinal < candidates[right].Ordinal
		}
		if candidates[left].ProjectID != candidates[right].ProjectID {
			return candidates[left].ProjectID < candidates[right].ProjectID
		}
		return candidates[left].WorkUnitID < candidates[right].WorkUnitID
	})
	progressed := false
	for _, candidate := range candidates {
		claimed, err := m.claimIntegration(ctx, candidate.Workspace)
		if err != nil {
			return progressed, err
		}
		if !claimed {
			var cmdID CommandID
			var wfRev WorkflowRevision
			if proj, ok := state.Projects[candidate.ProjectID]; ok && proj.Intent != nil {
				cmdID = proj.Intent.CommandID
				wfRev = proj.Intent.WorkflowRevision
			}
			m.emit(ctx, Event{
				Type: EventBlocked, ProjectID: candidate.ProjectID,
				WorkspaceID: candidate.Workspace, WorkUnitID: candidate.WorkUnitID,
				CommandID: cmdID, WorkflowRevision: wfRev,
				Detail: string(BlockerDuplicateWorkspaceCall),
			})
			continue
		}
		err = m.integrateOne(ctx, candidate)
		m.releaseIntegration(context.WithoutCancel(ctx), candidate.Workspace)
		if err != nil {
			return progressed, err
		}
		progressed = true
	}
	return progressed, nil
}

func (m *Manager) integrateOne(
	ctx context.Context,
	candidate integrationCandidate,
) error {
	project, work, err := m.loadWork(ctx, candidate.ProjectID, candidate.WorkUnitID)
	if err != nil {
		return err
	}
	if work.State != WorkStatePendingIntegration || work.ChangeSet == nil {
		return nil
	}
	integrationAttempt := work.IntegrationAttempt
	if integrationAttempt == 0 {
		integrationAttempt = 1
	}
	if err := m.changeWork(ctx, candidate.ProjectID, candidate.WorkUnitID, func(work *WorkRecord) error {
		if err := transitionWork(work, WorkStateIntegrating); err != nil {
			return err
		}
		work.IntegrationAttempt = integrationAttempt
		return nil
	}); err != nil {
		return err
	}
	project, work, err = m.loadWork(ctx, candidate.ProjectID, candidate.WorkUnitID)
	if err != nil {
		return err
	}
	request := IntegrationRequest{
		Project: project, Work: work, ChangeSet: *work.ChangeSet,
		Ordinal: work.DispatchOrdinal, Attempt: integrationAttempt,
		IdempotencyKey: integrationKey(
			candidate.ProjectID, candidate.WorkUnitID, *work.ChangeSet, integrationAttempt,
		),
	}
	result, integrateErr := m.integrator.Integrate(ctx, request)
	if integrateErr != nil {
		result = IntegrationResult{
			ProjectID: candidate.ProjectID, WorkUnitID: candidate.WorkUnitID,
			ChangeSet: *work.ChangeSet, Ordinal: work.DispatchOrdinal,
			Attempt: integrationAttempt, Outcome: IntegrationOutcomeTerminalDeferred,
			Retained: true,
			Blocker: &Blocker{
				Code: BlockerIntegrationFailed, Message: integrateErr.Error(), Retryable: true,
			},
		}
	}
	if err := validateIntegrationResult(request, result); err != nil {
		result = IntegrationResult{
			ProjectID: candidate.ProjectID, WorkUnitID: candidate.WorkUnitID,
			ChangeSet: *work.ChangeSet, Ordinal: work.DispatchOrdinal,
			Attempt: integrationAttempt, Outcome: IntegrationOutcomeTerminalDeferred,
			Retained: true,
			Blocker:  &Blocker{Code: BlockerIntegrationFailed, Message: err.Error()},
		}
	}
	var next WorkState
	switch result.Outcome {
	case IntegrationOutcomeIntegrated:
		next = WorkStateCompleted
	case IntegrationOutcomeTerminalDeferred:
		next = WorkStateTerminalDeferred
	default:
		return fmt.Errorf("agenttask: unsupported integration outcome %q", result.Outcome)
	}
	err = m.changeWork(ctx, candidate.ProjectID, candidate.WorkUnitID, func(work *WorkRecord) error {
		if err := transitionWork(work, next); err != nil {
			return err
		}
		work.Integration = &result
		work.Blocker = result.Blocker
		return nil
	})
	if err != nil {
		return err
	}
	var cmdID CommandID
	var wfRev WorkflowRevision
	if project.Intent != nil {
		cmdID = project.Intent.CommandID
		wfRev = project.Intent.WorkflowRevision
	}
	m.emit(ctx, Event{
		Type:               EventIntegrationResult,
		ProjectID:          candidate.ProjectID,
		WorkspaceID:        candidate.Workspace,
		WorkUnitID:         candidate.WorkUnitID,
		CommandID:          cmdID,
		WorkflowRevision:   wfRev,
		AttemptID:          work.AttemptID,
		Ordinal:            candidate.Ordinal,
		ChangeSetID:        result.ChangeSet.ID,
		ChangeSetRevision:  result.ChangeSet.Revision,
		IntegrationAttempt: result.Attempt,
		State:              next,
		Detail:             string(result.Outcome),
	})
	return nil
}

func validateIntegrationResult(request IntegrationRequest, result IntegrationResult) error {
	if result.ProjectID != request.Project.ProjectID ||
		result.WorkUnitID != request.Work.Unit.ID ||
		result.Ordinal != request.Ordinal ||
		result.Attempt != request.Attempt ||
		!reflect.DeepEqual(result.ChangeSet, request.ChangeSet) {
		return fmt.Errorf("agenttask: integration result durable identity mismatch")
	}
	switch result.Outcome {
	case IntegrationOutcomeIntegrated:
		if result.Blocker != nil {
			return fmt.Errorf("agenttask: integrated result cannot contain a blocker")
		}
	case IntegrationOutcomeTerminalDeferred:
		if !result.Retained || result.Blocker == nil {
			return fmt.Errorf("agenttask: terminal-deferred integration must retain its change set and blocker")
		}
	default:
		return fmt.Errorf("agenttask: unknown integration outcome")
	}
	return nil
}