fix: 隔离 Pod 与 VM 并发容量

cmd/gitea-dynamic-runner/controller.go

@@ -15,6 +15,11 @@ import (
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 	"golang.org/x/sync/errgroup"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
 
 	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/assignmentqueue"
 	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/controller"
@@ -86,7 +91,6 @@ func runController(ctx context.Context) error {
 		return err
 	}
 	registry := runnerfacade.NewRegistry()
-	gate := taskscheduler.NewSingleFlightGate()
 	var podExecutorBackend *podbackend.Backend
 	var vmExecutorBackend *opensandboxbackend.Backend
 	giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken)
@@ -109,7 +113,6 @@ func runController(ctx context.Context) error {
 					return err
 				}
 			}
-			gate.Release()
 			return nil
 		},
 	}
@@ -126,13 +129,25 @@ func runController(ctx context.Context) error {
 		labels = append(labels, string(taskassignment.BackendVM))
 	}
 	poller := taskscheduler.Poller{
-		Client: giteaClient, Gate: gate,
+		Client: giteaClient,
 		Scheduler: &taskscheduler.Scheduler{TrustDomain: config.TrustDomain, Dispatcher: assignmentqueue.Publisher{
 			JetStream: producerJS, SubjectBase: config.SubjectBase,
 		}},
 		Config:  taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels},
 		OnError: func(err error) { log.Printf("scheduler: %v", err) },
 	}
+	kubernetesConfig, err := rest.InClusterConfig()
+	if err != nil {
+		return fmt.Errorf("load leader election Kubernetes config: %w", err)
+	}
+	kubernetesClient, err := kubernetes.NewForConfig(kubernetesConfig)
+	if err != nil {
+		return fmt.Errorf("create leader election Kubernetes client: %w", err)
+	}
+	leaderIdentity := strings.TrimSpace(os.Getenv("HOSTNAME"))
+	if leaderIdentity == "" {
+		return errors.New("HOSTNAME is required for scheduler leader election")
+	}
 	facadeServer := runnerfacade.Server{
 		Facade: facade, ListenAddress: config.FacadeListen, TrustDomain: config.TrustDomain,
 		WorkloadAPIAddr: config.WorkloadAPIAddr, UpstreamURL: config.GiteaURL,
@@ -141,7 +156,9 @@ func runController(ctx context.Context) error {
 		controller.Scheduler: runComponent(func(ctx context.Context) error {
 			group, groupContext := errgroup.WithContext(ctx)
 			group.Go(func() error { return facadeServer.Run(groupContext) })
-			group.Go(func() error { return poller.Run(groupContext) })
+			group.Go(func() error {
+				return runSchedulerLeader(groupContext, kubernetesClient, config.PodNamespace, leaderIdentity, poller.Run)
+			})
 			return group.Wait()
 		}),
 	}
@@ -158,6 +175,15 @@ func runController(ctx context.Context) error {
 			SPIREAgentID: config.SPIREAgentID, ExecutorUID: config.PodExecutorUID,
 		}}
 		podExecutorBackend = &backend
+		assignments, err := backend.RecoverAssignments(ctx)
+		if err != nil {
+			return err
+		}
+		for _, assignment := range assignments {
+			if err := registry.RecoverClaimed(assignment); err != nil {
+				return fmt.Errorf("recover Pod facade claim %s: %w", assignment.ID, err)
+			}
+		}
 		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry)
 		if err != nil {
 			return err
@@ -178,6 +204,15 @@ func runController(ctx context.Context) error {
 			Env:        map[string]string{"SPIFFE_ENDPOINT_SOCKET": config.WorkloadAPIAddr},
 		}}
 		vmExecutorBackend = &backend
+		assignments, err := backend.RecoverAssignments(ctx, config.TrustDomain)
+		if err != nil {
+			return err
+		}
+		for _, assignment := range assignments {
+			if err := registry.RecoverClaimed(assignment); err != nil {
+				return fmt.Errorf("recover VM facade claim %s: %w", assignment.ID, err)
+			}
+		}
 		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry)
 		if err != nil {
 			return err
@@ -193,6 +228,50 @@ func runController(ctx context.Context) error {
 	return controller.Run(ctx, config.Components, components)
 }
 
+func runSchedulerLeader(ctx context.Context, client kubernetes.Interface, namespace, identity string, run func(context.Context) error) error {
+	if client == nil || namespace == "" || identity == "" || run == nil {
+		return errors.New("leader election client, namespace, identity, and scheduler are required")
+	}
+	electionContext, cancel := context.WithCancel(ctx)
+	defer cancel()
+	result := make(chan error, 1)
+	lock := &resourcelock.LeaseLock{
+		LeaseMeta: metav1.ObjectMeta{Name: "dynamic-runner-scheduler", Namespace: namespace},
+		Client:    client.CoordinationV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity: identity,
+		},
+	}
+	elector, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
+		Lock: lock, LeaseDuration: 15 * time.Second, RenewDeadline: 10 * time.Second, RetryPeriod: 2 * time.Second,
+		ReleaseOnCancel: true,
+		Callbacks: leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(leaderContext context.Context) {
+				result <- run(leaderContext)
+				cancel()
+			},
+			OnStoppedLeading: func() {
+				if ctx.Err() == nil {
+					select {
+					case result <- errors.New("scheduler leadership lost"):
+					default:
+					}
+				}
+			},
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("configure scheduler leader election: %w", err)
+	}
+	go elector.Run(electionContext)
+	select {
+	case err := <-result:
+		return err
+	case <-ctx.Done():
+		return nil
+	}
+}
+
 func connectNATS(server, user, password, caFile, clientName string) (*nats.Conn, error) {
 	options := []nats.Option{nats.Name(clientName), nats.UserInfo(user, password)}
 	if caFile != "" {

docs/runner-protocol-roadmap.md

@@ -80,9 +80,9 @@ facade，并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把
 - consumer 在 executor 使用上述 facade 成功 claim task 后确认 assignment；无需把完整
   task 写入 Pod annotation、OpenSandbox metadata 或环境变量。
 - Pod 与 VM 共享 task/executor 协议，只有环境创建和销毁实现不同。
-- 首次生产 canary 使用 controller 进程内 single-flight gate：只有官方 runner 的终态
-  `UpdateTask` 已被 Gitea 接受后才允许 Fetch 下一条任务。它把未知故障收敛为停止领取，
-  而不是在 backlog 下连续创建 executor；后续容量调度必须以 backend 实际运行资源为准。
+- scheduler 在 assignment 持久化到 JetStream 后即可继续领取；Pod 与 VM 分别由 durable
+  consumer 的 capacity 限制并发，不共享全局执行槽位。未知后端故障由对应 consumer 的
+  NAK/redelivery 收敛，不能阻塞另一种 backend。
 - 两种 backend 都注入同一份 runner bootstrap 环境；Pod 仍由 homelab Kubernetes 原生
   创建，只有 VM 经 OpenSandbox 创建，bootstrap 机制不改变 backend 边界。
 

go.mod

@@ -45,6 +45,7 @@ require (
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/nats-io/nkeys v0.4.16 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/sirupsen/logrus v1.10.2 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	go.yaml.in/yaml/v2 v2.4.4 // indirect

go.sum

@@ -112,6 +112,8 @@ go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2W
 go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew=
 go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI=
 go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
 go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
 go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw=

internal/opensandboxbackend/backend.go

@@ -108,6 +108,30 @@ type Backend struct {
 	Config    Config
 }
 
+func (b Backend) RecoverAssignments(ctx context.Context, trustDomain string) ([]taskassignment.Assignment, error) {
+	if err := b.validate(); err != nil {
+		return nil, err
+	}
+	result, err := b.Lifecycle.ListSandboxes(ctx, opensandbox.ListOptions{
+		Metadata: map[string]string{"ci.ddupan.top/backend": "vm"}, PageSize: 100,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("list recoverable sandboxes: %w", err)
+	}
+	assignments := make([]taskassignment.Assignment, 0, len(result.Items))
+	for _, sandbox := range result.Items {
+		if sandbox.Metadata[terminalMetadata] == "true" {
+			continue
+		}
+		assignment, err := taskassignment.FromMetadata(sandbox.Metadata, sandbox.Metadata, trustDomain)
+		if err != nil {
+			return nil, fmt.Errorf("recover sandbox %s: %w", sandbox.ID, err)
+		}
+		assignments = append(assignments, assignment)
+	}
+	return assignments, nil
+}
+
 func NewLifecycleClient(baseURL, apiKey string, client *http.Client) *opensandbox.LifecycleClient {
 	if client != nil {
 		return opensandbox.NewLifecycleClient(baseURL, apiKey, opensandbox.WithHTTPClient(client))

internal/podbackend/backend.go

@@ -116,6 +116,28 @@ type Backend struct {
 	Config Config
 }
 
+func (b Backend) RecoverAssignments(ctx context.Context) ([]taskassignment.Assignment, error) {
+	if err := b.validate(); err != nil {
+		return nil, err
+	}
+	pods, err := b.API.ListPods(ctx, b.Config.Namespace, "ci.ddupan.top/backend=pod")
+	if err != nil {
+		return nil, fmt.Errorf("list recoverable assignment Pods: %w", err)
+	}
+	assignments := make([]taskassignment.Assignment, 0, len(pods))
+	for _, pod := range pods {
+		if pod.Labels[terminalLabel] == "true" {
+			continue
+		}
+		assignment, err := taskassignment.FromMetadata(pod.Labels, pod.Annotations, b.Config.TrustDomain)
+		if err != nil {
+			return nil, fmt.Errorf("recover Pod %s: %w", pod.Name, err)
+		}
+		assignments = append(assignments, assignment)
+	}
+	return assignments, nil
+}
+
 func (b Backend) Find(ctx context.Context, assignmentID string) (*taskworker.Executor, error) {
 	if err := b.validate(); err != nil {
 		return nil, err

internal/runnerfacade/facade_test.go

@@ -225,3 +225,15 @@ func TestCapabilitiesAreDeterministicAndAssignmentScoped(t *testing.T) {
 		t.Fatal("capability scope is invalid")
 	}
 }
+
+func TestRegistryRecoversClaimedAssignment(t *testing.T) {
+	registry := NewRegistry()
+	assignment := facadeAssignment(t)
+	if err := registry.RecoverClaimed(assignment); err != nil {
+		t.Fatal(err)
+	}
+	resolved, err := registry.Resolve(assignment.ID, assignment.Identity.SPIFFEID)
+	if err != nil || resolved.Task.GetId() != assignment.Task.GetId() {
+		t.Fatalf("resolved=%#v err=%v", resolved, err)
+	}
+}

internal/runnerfacade/registry.go

@@ -90,3 +90,21 @@ func (r *Registry) Remove(assignmentID string) {
 	defer r.mu.Unlock()
 	delete(r.claims, assignmentID)
 }
+
+// RecoverClaimed restores authorization for an executor that already claimed
+// its task before the controller restarted.
+func (r *Registry) RecoverClaimed(assignment taskassignment.Assignment) error {
+	ready, err := r.Offer(assignment)
+	if err != nil {
+		return err
+	}
+	r.mu.Lock()
+	entry := r.claims[assignment.ID]
+	if !entry.claimed {
+		entry.claimed = true
+		close(entry.ready)
+	}
+	r.mu.Unlock()
+	<-ready
+	return nil
+}

internal/taskassignment/assignment.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"slices"
+	"strconv"
 
 	"gitea.dev/actionslib/pkg/model"
 	runnerv1 "gitea.dev/actionslib/runner/v1"
@@ -32,6 +33,32 @@ type Assignment struct {
 	Identity taskidentity.Identity
 }
 
+// FromMetadata reconstructs the minimal assignment needed to authorize an
+// already-running executor after a controller restart. Backend 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")
+	}
+	backend := Backend(labels["ci.ddupan.top/backend"])
+	if backend != BackendPod && backend != BackendVM {
+		return Assignment{}, errors.New("backend metadata has invalid backend")
+	}
+	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, Backend: backend, Task: &runnerv1.Task{Id: taskID}, Identity: identity}, nil
+}
+
 type envelope struct {
 	Version  int                   `json:"version"`
 	ID       string                `json:"id"`

internal/taskassignment/assignment_test.go

@@ -73,3 +73,18 @@ func TestAssignmentWireRoundTripAndValidation(t *testing.T) {
 		t.Fatal("expected tampered backend to fail")
 	}
 }
+
+func TestFromMetadataRecoversMinimalAssignment(t *testing.T) {
+	assignment, err := FromMetadata(map[string]string{
+		"ci.ddupan.top/assignment-id": "gitea-task-42",
+		"ci.ddupan.top/task-id":       "42",
+		"ci.ddupan.top/backend":       "vm",
+	}, map[string]string{
+		"ci.ddupan.top/repository": "owner/repo",
+		"ci.ddupan.top/job-key":    "publish",
+		"ci.ddupan.top/spiffe-id":  "spiffe://ddupan.top/ci/owner/repo/publish",
+	}, "ddupan.top")
+	if err != nil || assignment.Task.GetId() != 42 || assignment.Backend != BackendVM {
+		t.Fatalf("assignment=%#v err=%v", assignment, err)
+	}
+}

internal/taskidentity/identity.go

@@ -23,6 +23,21 @@ type Identity struct {
 	SPIFFEID   string
 }
 
+// FromMetadata validates identity fields recovered from backend-owned state.
+func FromMetadata(repository, task, spiffeID, trustDomain string) (Identity, error) {
+	parts := strings.Split(repository, "/")
+	if len(parts) != 2 || parts[0] == "" || parts[1] == "" || !safeTaskKey.MatchString(task) {
+		return Identity{}, errors.New("invalid recovered repository or task identity")
+	}
+	expected := "spiffe://" + trustDomain + "/ci/" + strings.Join([]string{
+		sanitize(parts[0]), sanitize(parts[1]), task,
+	}, "/")
+	if spiffeID != expected {
+		return Identity{}, fmt.Errorf("recovered SPIFFE ID %q does not match %q", spiffeID, expected)
+	}
+	return Identity{Repository: repository, Task: task, SPIFFEID: spiffeID}, nil
+}
+
 // FromTask derives the repository/task SPIFFE ID from Gitea's trusted task
 // context. Workflow input never supplies or overrides the resulting ID.
 func FromTask(task *runnerv1.Task, trustDomain string) (Identity, error) {

internal/taskscheduler/gate.go

@@ -1,31 +0,0 @@
-package taskscheduler
-
-import "context"
-
-// SingleFlightGate keeps at most one fetched task in flight. Release is
-// idempotent so repeated terminal updates cannot increase capacity.
-type SingleFlightGate struct {
-	token chan struct{}
-}
-
-func NewSingleFlightGate() *SingleFlightGate {
-	gate := &SingleFlightGate{token: make(chan struct{}, 1)}
-	gate.token <- struct{}{}
-	return gate
-}
-
-func (g *SingleFlightGate) Acquire(ctx context.Context) error {
-	select {
-	case <-g.token:
-		return nil
-	case <-ctx.Done():
-		return ctx.Err()
-	}
-}
-
-func (g *SingleFlightGate) Release() {
-	select {
-	case g.token <- struct{}{}:
-	default:
-	}
-}

internal/taskscheduler/gate_test.go

@@ -1,29 +0,0 @@
-package taskscheduler
-
-import (
-	"context"
-	"testing"
-	"time"
-)
-
-func TestSingleFlightGateBlocksUntilTerminalRelease(t *testing.T) {
-	gate := NewSingleFlightGate()
-	if err := gate.Acquire(context.Background()); err != nil {
-		t.Fatal(err)
-	}
-	blocked, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
-	defer cancel()
-	if err := gate.Acquire(blocked); err == nil {
-		t.Fatal("second task acquired capacity before release")
-	}
-	gate.Release()
-	gate.Release()
-	if err := gate.Acquire(context.Background()); err != nil {
-		t.Fatal(err)
-	}
-	blockedAgain, cancelAgain := context.WithTimeout(context.Background(), 20*time.Millisecond)
-	defer cancelAgain()
-	if err := gate.Acquire(blockedAgain); err == nil {
-		t.Fatal("duplicate terminal release increased capacity")
-	}
-}

internal/taskscheduler/poller.go

@@ -26,7 +26,6 @@ type PollerConfig struct {
 type Poller struct {
 	Client    PollClient
 	Scheduler *Scheduler
-	Gate      *SingleFlightGate
 	Config    PollerConfig
 	OnError   func(error)
 }
@@ -51,14 +50,7 @@ func (p Poller) Run(ctx context.Context) error {
 		errorBackoff = 5 * time.Second
 	}
 	var tasksVersion int64
-	haveLease := false
 	for {
-		if p.Gate != nil && !haveLease {
-			if err := p.Gate.Acquire(ctx); err != nil {
-				return nil
-			}
-			haveLease = true
-		}
 		response, err := p.Client.FetchTask(ctx, tasksVersion)
 		if err != nil {
 			if ctx.Err() != nil {
@@ -80,10 +72,6 @@ func (p Poller) Run(ctx context.Context) error {
 		tasksVersion = response.GetTasksVersion()
 		task := response.GetTask()
 		if task == nil {
-			if p.Gate != nil {
-				p.Gate.Release()
-				haveLease = false
-			}
 			if !wait(ctx, emptyBackoff) {
 				return nil
 			}
@@ -91,7 +79,6 @@ func (p Poller) Run(ctx context.Context) error {
 		}
 		for {
 			if err := p.Scheduler.Run(ctx, task); err == nil {
-				haveLease = false
 				break
 			} else {
 				if ctx.Err() != nil {