// 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 } // 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.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 } 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/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, }, } }