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,
	leases *leaseSet,
) (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 {
		if !integrationCandidateReady(state, candidate) {
			continue
		}
		integrationClaim, err := m.claimIntegration(ctx, candidate.Workspace)
		if err != nil {
			return progressed, err
		}
		if integrationClaim == nil {
			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
		}
		leases.Add(integrationClaim)
		err = m.integrateOne(ctx, candidate, integrationClaim)
		// Remove first so an in-flight supervisor pass cannot renew a claim
		// after its exact-token release. Remove waits for such a pass to finish.
		leases.Remove(integrationClaim)
		m.releaseExact(unfencedLeaseContext(ctx), integrationClaim)
		if err != nil {
			return progressed, err
		}
		progressed = true
		state, err = m.load(ctx)
		if err != nil {
			return progressed, err
		}
	}
	return progressed, nil
}

func (m *Manager) integrateOne(
	ctx context.Context,
	candidate integrationCandidate,
	integrationClaim *leaseClaim,
) 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 := nextIntegrationAttempt(work)
	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)
	}

	// Fence: reject late results that arrived after ownership was lost.
	if fenceErr := m.validateIntegration(ctx, integrationClaim.token, integrationClaim.subject); fenceErr != nil {
		return fenceErr
	}

	err = m.changeWork(ctx, candidate.ProjectID, candidate.WorkUnitID, func(work *WorkRecord) error {
		if err := transitionWork(work, next); err != nil {
			return err
		}
		work.Integration = &result
		if result.Outcome == IntegrationOutcomeIntegrated {
			work.CompletionVerified = true
			resetFailure(work, FailureStageIntegration)
		} else if result.Blocker != nil {
			blocker := m.recordFailure(work, FailureStageIntegration, *result.Blocker)
			work.Blocker = &blocker
		}
		if result.CompletionLocator != nil {
			if work.Locators == nil {
				work.Locators = make(map[LocatorKind]LocatorRecord)
			}
			work.Locators[LocatorCompletion] = *result.CompletionLocator
		}
		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 integrationCandidateReady(
	state ManagerState,
	candidate integrationCandidate,
) bool {
	for _, project := range state.Projects {
		if project.WorkspaceID != candidate.Workspace {
			continue
		}
		for _, work := range project.Works {
			if work.DispatchOrdinal == 0 ||
				work.DispatchOrdinal >= candidate.Ordinal {
				continue
			}
			// Only a terminal disposition releases the queue. Blocked, stopped,
			// reviewing, and pending work can still return to ready under the
			// same dispatch ordinal, so they remain barriers even when they
			// currently hold no change set.
			if work.State == WorkStateCompleted ||
				work.State == WorkStateTerminalDeferred {
				continue
			}
			return false
		}
	}
	return true
}

func nextIntegrationAttempt(work WorkRecord) IntegrationAttempt {
	if work.IntegrationAttempt == 0 {
		return 1
	}
	if work.Integration != nil &&
		!reflect.DeepEqual(work.Integration.ChangeSet, *work.ChangeSet) {
		return work.IntegrationAttempt + 1
	}
	return work.IntegrationAttempt
}

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")
	}
	if result.CompletionLocator != nil {
		if err := validateLocator(request.Project, request.Work, *result.CompletionLocator); err != nil {
			return err
		}
		if result.CompletionLocator.Kind != LocatorCompletion {
			return fmt.Errorf("agenttask: integration completion locator has kind %q", result.CompletionLocator.Kind)
		}
	}
	return nil
}