gitea-dynamic-runner/internal/taskassignment/assignment.go

// Package taskassignment defines the durable handoff between the scheduler and workers.
package taskassignment

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"

	"gitea.dev/actionslib/pkg/model"
	runnerv1 "gitea.dev/actionslib/runner/v1"
	"google.golang.org/protobuf/proto"

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

const wireVersion = 2

// Assignment is the only document persisted in the handoff queue.
type Assignment struct {
	ID        string
	Placement Placement
	Task      *runnerv1.Task
	Identity  taskidentity.Identity
}

// FromMetadata reconstructs the minimal assignment needed to authorize an
// already-running executor after a controller restart. Placement metadata was
// originally derived from the trusted Gitea task and is validated again here.
func FromMetadata(labels, annotations map[string]string, trustDomain string) (Assignment, error) {
	taskID, err := strconv.ParseInt(labels["ci.ddupan.top/task-id"], 10, 64)
	if err != nil || taskID < 1 {
		return Assignment{}, errors.New("backend metadata has invalid task ID")
	}
	placement := Placement{Class: WorkloadClass(labels["ci.ddupan.top/workload-class"]), Driver: Driver(labels["ci.ddupan.top/driver"])}
	if err := placement.Validate(); err != nil {
		return Assignment{}, err
	}
	id := labels["ci.ddupan.top/assignment-id"]
	if id != fmt.Sprintf("gitea-task-%d", taskID) {
		return Assignment{}, errors.New("backend metadata assignment ID does not match task ID")
	}
	identity, err := taskidentity.FromMetadata(
		annotations["ci.ddupan.top/repository"], annotations["ci.ddupan.top/job-key"],
		annotations["ci.ddupan.top/spiffe-id"], trustDomain,
	)
	if err != nil {
		return Assignment{}, err
	}
	return Assignment{ID: id, Placement: placement, Task: &runnerv1.Task{Id: taskID}, Identity: identity}, nil
}

type envelope struct {
	Version   int                   `json:"version"`
	ID        string                `json:"id"`
	Placement Placement             `json:"placement"`
	Task      []byte                `json:"task"`
	Identity  taskidentity.Identity `json:"identity"`
}

// New derives all trusted assignment fields from the task fetched from Gitea.
func New(task *runnerv1.Task, trustDomain string) (Assignment, error) {
	if task == nil || task.GetId() <= 0 {
		return Assignment{}, errors.New("positive Gitea task ID is required")
	}
	identity, err := taskidentity.FromTask(task, trustDomain)
	if err != nil {
		return Assignment{}, err
	}
	placement, err := placementFromTask(task)
	if err != nil {
		return Assignment{}, err
	}
	return Assignment{
		ID:        fmt.Sprintf("gitea-task-%d", task.GetId()),
		Placement: placement,
		Task:      task,
		Identity:  identity,
	}, nil
}

func placementFromTask(task *runnerv1.Task) (Placement, error) {
	workflow, err := model.ReadWorkflow(bytes.NewReader(task.GetWorkflowPayload()))
	if err != nil {
		return Placement{}, fmt.Errorf("parse task workflow for placement: %w", err)
	}
	jobIDs := workflow.GetJobIDs()
	if len(jobIDs) != 1 || workflow.GetJob(jobIDs[0]) == nil {
		return Placement{}, fmt.Errorf("task workflow must contain exactly one non-empty job")
	}
	return PlacementFromLabels(workflow.GetJob(jobIDs[0]).RunsOnLabels())
}

// Marshal encodes a versioned assignment. Protobuf preserves the exact Gitea task.
func Marshal(assignment Assignment) ([]byte, error) {
	if assignment.Task == nil {
		return nil, errors.New("assignment task is required")
	}
	if err := assignment.Placement.Validate(); err != nil {
		return nil, err
	}
	task, err := proto.Marshal(assignment.Task)
	if err != nil {
		return nil, fmt.Errorf("marshal Gitea task: %w", err)
	}
	return json.Marshal(envelope{
		Version: wireVersion,
		ID:      assignment.ID, Placement: assignment.Placement,
		Task: task, Identity: assignment.Identity,
	})
}

// Unmarshal re-derives trusted fields instead of trusting duplicated queue metadata.
func Unmarshal(data []byte, trustDomain string) (Assignment, error) {
	var wire envelope
	if err := json.Unmarshal(data, &wire); err != nil {
		return Assignment{}, fmt.Errorf("decode assignment: %w", err)
	}
	if wire.Version != wireVersion {
		return Assignment{}, fmt.Errorf("unsupported assignment version %d", wire.Version)
	}
	task := new(runnerv1.Task)
	if err := proto.Unmarshal(wire.Task, task); err != nil {
		return Assignment{}, fmt.Errorf("unmarshal Gitea task: %w", err)
	}
	canonical, err := New(task, trustDomain)
	if err != nil {
		return Assignment{}, err
	}
	if wire.ID != canonical.ID || wire.Placement != canonical.Placement || wire.Identity != canonical.Identity {
		return Assignment{}, errors.New("assignment metadata does not match its Gitea task")
	}
	return canonical, nil
}