gitea-dynamic-runner/internal/taskworker/worker.go

// Package taskworker reconciles assigned Gitea tasks against an executor backend.
package taskworker

import (
	"context"
	"errors"
	"strconv"

	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
)

// Phase is observed from Kubernetes or OpenSandbox, never stored by the worker.
type Phase string

const (
	PhasePending   Phase = "pending"
	PhaseRunning   Phase = "running"
	PhaseSucceeded Phase = "succeeded"
	PhaseFailed    Phase = "failed"
)

// Executor is the backend resource discovered by assignment ID.
type Executor struct {
	Name           string
	IdentityTarget string
	Phase          Phase
}

// Metadata is persisted on the backend resource. Labels remain query-safe;
// annotations retain complete identity context.
type Metadata struct {
	Labels      map[string]string
	Annotations map[string]string
}

// LaunchSpec contains the durable resource metadata and the short-lived
// executor environment. Environment values configure the one-shot runner but
// are deliberately excluded from labels and annotations.
type LaunchSpec struct {
	Metadata    Metadata
	Environment map[string]string
}

// Bootstrap produces assignment-scoped executor configuration. Implementations
// must be deterministic so a redelivery after controller restart creates the
// same credentials without storing another lifecycle record.
type Bootstrap interface {
	Environment(taskassignment.Assignment) (map[string]string, error)
}

// Backend is implemented by the native Pod and OpenSandbox adapters.
// Every method must be idempotent.
type Backend interface {
	Find(context.Context, string) (*Executor, error)
	Create(context.Context, taskassignment.Assignment, LaunchSpec) (*Executor, error)
	BindIdentity(context.Context, *Executor, taskidentity.Identity) error
	Delete(context.Context, *Executor) error
}

// TaskState uses Gitea as the authority for whether an assigned task has
// already reached a terminal state.
type TaskState interface {
	Terminal(context.Context, int64) (bool, error)
	Report(context.Context, int64, Phase) error
}

// Worker has no correctness-critical in-memory state. Handle may be called
// again for the same assignment after any operation.
type Worker struct {
	Backend   Backend
	Tasks     TaskState
	Bootstrap Bootstrap
	OnEvent   func(Event)
}

// Event describes a backend lifecycle transition without exposing launch
// environment values or other credentials.
type Event struct {
	Name         string
	AssignmentID string
	Executor     string
	Phase        Phase
}

func (w Worker) event(name string, assignment taskassignment.Assignment, executor *Executor) {
	if w.OnEvent == nil {
		return
	}
	event := Event{Name: name, AssignmentID: assignment.ID}
	if executor != nil {
		event.Executor = executor.Name
		event.Phase = executor.Phase
	}
	w.OnEvent(event)
}

// Accept completes the durable handoff from JetStream to the backend. Once it
// returns true, all recovery information exists in Kubernetes/OpenSandbox and
// the assignment message can be acknowledged immediately.
func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment) (bool, error) {
	if w.Backend == nil || w.Bootstrap == nil {
		return false, errors.New("backend and runner bootstrap are required")
	}
	if assignment.ID == "" || assignment.Task == nil {
		return false, errors.New("valid assignment is required")
	}
	executor, err := w.Backend.Find(ctx, assignment.ID)
	if err != nil {
		return false, err
	}
	if executor == nil {
		w.event("executor_absent", assignment, nil)
		launch, launchErr := w.launchSpec(assignment)
		if launchErr != nil {
			return false, launchErr
		}
		executor, err = w.Backend.Create(ctx, assignment, launch)
		if err != nil {
			return false, err
		}
		w.event("executor_created", assignment, executor)
	} else {
		w.event("executor_found", assignment, executor)
	}
	if executor.IdentityTarget == "" {
		w.event("identity_target_pending", assignment, executor)
		return false, nil
	}
	if err := w.Backend.BindIdentity(ctx, executor, assignment.Identity); err != nil {
		return false, err
	}
	w.event("identity_bound", assignment, executor)
	return true, nil
}

// Handle performs one reconciliation. Done means the queue message may be
// acknowledged. A false result should remain pending and be reconciled again.
func (w Worker) Handle(ctx context.Context, assignment taskassignment.Assignment) (done bool, err error) {
	if w.Backend == nil || w.Tasks == nil || w.Bootstrap == nil {
		return false, errors.New("backend, Gitea task state, and runner bootstrap are required")
	}
	if assignment.ID == "" || assignment.Task == nil {
		return false, errors.New("valid assignment is required")
	}

	terminal, err := w.Tasks.Terminal(ctx, assignment.Task.GetId())
	if err != nil {
		return false, err
	}
	executor, err := w.Backend.Find(ctx, assignment.ID)
	if err != nil {
		return false, err
	}
	if terminal {
		if executor != nil {
			if err := w.Backend.Delete(ctx, executor); err != nil {
				return false, err
			}
		}
		return true, nil
	}

	if executor == nil {
		launch, launchErr := w.launchSpec(assignment)
		if launchErr != nil {
			return false, launchErr
		}
		executor, err = w.Backend.Create(ctx, assignment, launch)
		if err != nil {
			return false, err
		}
	}
	if executor.IdentityTarget == "" {
		return false, nil
	}
	if err := w.Backend.BindIdentity(ctx, executor, assignment.Identity); err != nil {
		return false, err
	}

	switch executor.Phase {
	case PhaseSucceeded, PhaseFailed:
		// Reporting first closes the delete-before-ack crash window: after a
		// restart Gitea prevents this assignment from executing a second time.
		if err := w.Tasks.Report(ctx, assignment.Task.GetId(), executor.Phase); err != nil {
			return false, err
		}
		if err := w.Backend.Delete(ctx, executor); err != nil {
			return false, err
		}
		return true, nil
	default:
		return false, nil
	}
}

func (w Worker) launchSpec(assignment taskassignment.Assignment) (LaunchSpec, error) {
	environment, err := w.Bootstrap.Environment(assignment)
	if err != nil {
		return LaunchSpec{}, err
	}
	return LaunchSpec{Metadata: BackendMetadata(assignment), Environment: environment}, nil
}

// BackendMetadata is the shared metadata contract for Pods and OpenSandbox.
func BackendMetadata(assignment taskassignment.Assignment) Metadata {
	return Metadata{
		Labels: map[string]string{
			"ci.ddupan.top/runner":        "true",
			"ci.ddupan.top/assignment-id": assignment.ID,
			"ci.ddupan.top/task-id":       strconv.FormatInt(assignment.Task.GetId(), 10),
			"ci.ddupan.top/backend":       string(assignment.Backend),
		},
		Annotations: map[string]string{
			"ci.ddupan.top/repository": assignment.Identity.Repository,
			"ci.ddupan.top/job-key":    assignment.Identity.Task,
			"ci.ddupan.top/spiffe-id":  assignment.Identity.SPIFFEID,
		},
	}
}