Merge Runner 无中断恢复与容量隔离

README.md

@@ -18,8 +18,7 @@ runs-on: [self-hosted, vm]
 目标 Go controller 组件：
 
 - `scheduler`：以常驻 Gitea RunnerService 身份直接领取 task，并把完整 assignment
-  持久化到 JetStream；一个 registration 下按配置启动多个并发 `FetchTask` goroutine，
+  持久化到 JetStream；同时提供仅允许 SPIFFE mTLS 的 RunnerService facade。
-  同时提供仅允许 SPIFFE mTLS 的 RunnerService facade。
 - `pod-worker`：直接在 homelab Kubernetes 创建一次性 Pod。
 - `vm-worker`：通过 OpenSandbox Lifecycle API 从 `ci-vm` Pool 创建 Kata microVM。
 - 三个组件默认在同一个 Go 进程启用。首轮集成期间不允许只启动 worker，因为 facade
@@ -38,9 +37,8 @@ runs-on: [self-hosted, vm]
   动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。
 
 Pod 路径由 homelab 集群中的 `pod-worker` 直接创建 Kubernetes Pod。OpenSandbox 只用于
-VM/Kata workload；两个 backend 使用独立 durable consumer 和独立容量池。assignment 根据
+VM/Kata workload；两个 backend 使用独立 durable consumer，任一执行层故障不会阻塞另一条
-`runs-on` 进入对应池，池满时留在 JetStream pending，不会创建超出容量的 workload；任一
+部署。长期 RunnerService 协议路线见
-执行层故障不会阻塞另一条部署。长期 RunnerService 协议路线见
 [`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
 
 ## 开发

cmd/gitea-dynamic-runner/controller.go

@@ -22,7 +22,6 @@ import (
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 
 	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/assignmentqueue"
-	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/backendpool"
 	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/controller"
 	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/giteaactions"
 	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/opensandboxbackend"
@@ -92,8 +91,6 @@ func runController(ctx context.Context) error {
 		return err
 	}
 	registry := runnerfacade.NewRegistry()
-	podPool := backendpool.New(config.PodCapacity)
-	vmPool := backendpool.New(config.VMCapacity)
 	var podExecutorBackend *podbackend.Backend
 	var vmExecutorBackend *opensandboxbackend.Backend
 	giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken)
@@ -108,7 +105,6 @@ func runController(ctx context.Context) error {
 				if err := podExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil {
 					return err
 				}
-				podPool.Release(assignment.ID)
 			case taskassignment.BackendVM:
 				if vmExecutorBackend == nil {
 					return errors.New("VM lifecycle backend is not configured")
@@ -116,7 +112,6 @@ func runController(ctx context.Context) error {
 				if err := vmExecutorBackend.MarkTerminal(ctx, assignment.ID); err != nil {
 					return err
 				}
-				vmPool.Release(assignment.ID)
 			}
 			return nil
 		},
@@ -138,7 +133,7 @@ func runController(ctx context.Context) error {
 		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, Capacity: config.PodCapacity + config.VMCapacity},
+		Config:  taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels},
 		OnError: func(err error) { log.Printf("scheduler: %v", err) },
 	}
 	kubernetesConfig, err := rest.InClusterConfig()
@@ -185,12 +180,11 @@ func runController(ctx context.Context) error {
 			return err
 		}
 		for _, assignment := range assignments {
-			podPool.Restore(assignment.ID)
 			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, podPool)
+		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry)
 		if err != nil {
 			return err
 		}
@@ -215,12 +209,11 @@ func runController(ctx context.Context) error {
 			return err
 		}
 		for _, assignment := range assignments {
-			vmPool.Restore(assignment.ID)
 			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, vmPool)
+		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry)
 		if err != nil {
 			return err
 		}
@@ -287,14 +280,14 @@ func connectNATS(server, user, password, caFile, clientName string) (*nats.Conn,
 	return nats.Connect(server, options...)
 }
 
-func workerComponent(ctx context.Context, js jetstream.JetStream, config controllerConfig, backend taskassignment.Backend, capacity int, accepter assignmentqueue.Accepter, claims assignmentqueue.Claims, admission assignmentqueue.Admission) (controller.Component, error) {
+func workerComponent(ctx context.Context, js jetstream.JetStream, config controllerConfig, backend taskassignment.Backend, capacity int, accepter assignmentqueue.Accepter, claims assignmentqueue.Claims) (controller.Component, error) {
 	consumer, err := assignmentqueue.OpenConsumer(ctx, js, config.Stream, config.SubjectBase, backend, capacity)
 	if err != nil {
 		return nil, err
 	}
 	return assignmentqueue.ConsumerComponent{
 		Consumer: consumer, Capacity: capacity,
-		Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims, Admission: admission},
+		Processor: assignmentqueue.Processor{TrustDomain: config.TrustDomain, Accepter: accepter, Claims: claims},
 		OnError:   func(err error) { log.Printf("%s worker: %v", backend, err) },
 	}, nil
 }

docs/runner-protocol-roadmap.md

@@ -3,9 +3,8 @@
 ## 目标
 
 长期形态不依赖 `workflow_job` webhook 发现工作。controller 本身作为 Gitea Runner
-协议客户端注册，并声明 `self-hosted`、`pod` 和 `vm` labels；单个 registration 内按总
+协议客户端注册，并声明 `self-hosted`、`pod` 和 `vm` labels；它只在后端存在可用容量
-配置容量启动多个 `FetchTask` goroutine，再将 task 按 `runs-on` 交给 Pod 或 VM 的独立
+时领取 task，然后将该 task 交给一个一次性 Pod 或 microVM 执行。
-容量池，由一次性 Pod 或 microVM 执行。
 
 ```text
 Gitea RunnerService
@@ -48,10 +47,7 @@ facade，并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把
 ## 设计约束
 
 - 对 workflow 的接口保持 `[self-hosted, pod]` 和 `[self-hosted, vm]` 不变。
-- scheduler 使用单一 Gitea runner UUID/token 和一个 `Declare`，不为并发槽位重复注册；
+- scheduler 在没有对应 backend 容量时不领取 task，避免本地形成不可控积压。
-  `POD_CAPACITY + VM_CAPACITY` 决定并发 `FetchTask` goroutine 数量。
-- task 领取并持久化后按 backend 进入独立 durable consumer；对应容量池已满时延迟 NAK，
-  assignment 保持 JetStream pending，且不得创建超出配置容量的 workload。
 - scheduler Declare 后使用 RunnerService 长轮询；一旦 FetchTask 返回已分配 task，在
   JetStream publish 成功前只重试该 assignment，不领取下一项。
 - 每个 executor 只执行一个 task，完成后销毁。
@@ -76,11 +72,9 @@ facade，并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把
 - executor 成功 claim 后 ACK assignment。Gitea 接受 terminal update 后，facade 在后端
   metadata 写入持久 terminal marker；backend reconciler 仅在执行环境也进入终态后清理，
   从而关闭进程重启窗口且避免删除尚未完成结果上报的环境。
-- pod 与 vm 使用独立 durable consumer、进程内 admission pool 和并发上限。consumer 只负责将 assignment
+- pod 与 vm 使用独立 durable consumer 和并发上限。consumer 只负责将 assignment
   幂等落到后端；executor 与身份恢复 metadata 持久化后立即 `DoubleAck`。尚未取得
   Pod UID 等短暂未就绪状态以及临时后端错误使用延迟 NAK。
-- admission pool 只保存可重建的并发状态：启动时从 Pod labels/annotations 或 OpenSandbox
-  metadata 恢复非终态 assignment，terminal update 持久化成功后释放槽位，不引入新存储。
 - assignment ACK 后的运行、结果回报和清理由 backend reconciler 根据 Kubernetes、
   OpenSandbox 与 Gitea 的事实状态驱动，不继续占用 JetStream delivery。
 - consumer 在 executor 使用上述 facade 成功 claim task 后确认 assignment；无需把完整

internal/assignmentqueue/jetstream.go

@@ -57,11 +57,6 @@ type Claims interface {
 	WaitClaimed(context.Context, string) error
 }
 
-type Admission interface {
-	Acquire(string) bool
-	Release(string)
-}
-
 // Message is the subset of jetstream.Msg needed by one reconciliation.
 type Message interface {
 	Data() []byte
@@ -75,14 +70,13 @@ type Processor struct {
 	TrustDomain  string
 	Accepter     Accepter
 	Claims       Claims
-	Admission    Admission
 	RetryDelay   time.Duration
 	ClaimTimeout time.Duration
 }
 
 func (p Processor) Process(ctx context.Context, message Message) error {
-	if p.Accepter == nil || p.Claims == nil || p.Admission == nil {
+	if p.Accepter == nil || p.Claims == nil {
-		return errors.New("assignment accepter, claim registry, and backend admission pool are required")
+		return errors.New("assignment accepter and claim registry are required")
 	}
 	assignment, err := taskassignment.Unmarshal(message.Data(), p.TrustDomain)
 	if err != nil {
@@ -91,16 +85,8 @@ func (p Processor) Process(ctx context.Context, message Message) error {
 	if _, err := p.Claims.Offer(assignment); err != nil {
 		return errors.Join(err, message.TermWithReason("conflicting assignment"))
 	}
-	if !p.Admission.Acquire(assignment.ID) {
-		delay := p.RetryDelay
-		if delay <= 0 {
-			delay = 2 * time.Second
-		}
-		return message.NakWithDelay(delay)
-	}
 	accepted, err := p.Accepter.Accept(ctx, assignment)
 	if err != nil {
-		p.Admission.Release(assignment.ID)
 		delay := p.RetryDelay
 		if delay <= 0 {
 			delay = 15 * time.Second

internal/assignmentqueue/jetstream_test.go

@@ -61,28 +61,6 @@ type fakeClaims struct {
 	claimed bool
 }
 
-type fakeAdmission struct {
-	allowed  bool
-	active   map[string]bool
-	released int
-}
-
-func (a *fakeAdmission) Acquire(assignmentID string) bool {
-	if !a.allowed {
-		return false
-	}
-	if a.active == nil {
-		a.active = make(map[string]bool)
-	}
-	a.active[assignmentID] = true
-	return true
-}
-
-func (a *fakeAdmission) Release(assignmentID string) {
-	delete(a.active, assignmentID)
-	a.released++
-}
-
 func (c *fakeClaims) Offer(taskassignment.Assignment) (<-chan struct{}, error) {
 	ready := make(chan struct{})
 	if c.claimed {
@@ -126,7 +104,7 @@ func encodedAssignment(t *testing.T) []byte {
 
 func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) {
 	message := &fakeMessage{data: encodedAssignment(t)}
-	processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}, Admission: &fakeAdmission{allowed: true}}
+	processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{accepted: true}, Claims: &fakeClaims{claimed: true}}
 	if err := processor.Process(context.Background(), message); err != nil {
 		t.Fatal(err)
 	}
@@ -137,7 +115,7 @@ func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) {
 
 func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) {
 	message := &fakeMessage{data: encodedAssignment(t)}
-	processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{}, Claims: &fakeClaims{}, Admission: &fakeAdmission{allowed: true}, RetryDelay: 2 * time.Second}
+	processor := Processor{TrustDomain: "ddupan.top", Accepter: &fakeAccepter{}, Claims: &fakeClaims{}, RetryDelay: 2 * time.Second}
 	if err := processor.Process(context.Background(), message); err != nil {
 		t.Fatal(err)
 	}
@@ -148,12 +126,10 @@ func TestProcessorRetriesUntilBackendHandoffIsDurable(t *testing.T) {
 
 func TestProcessorRetriesBackendFailureAndTerminatesPoisonMessage(t *testing.T) {
 	retry := &fakeMessage{data: encodedAssignment(t)}
-	admission := &fakeAdmission{allowed: true}
 	processor := Processor{
 		TrustDomain: "ddupan.top",
 		Accepter:    &fakeAccepter{err: errors.New("backend unavailable")},
 		Claims:      &fakeClaims{},
-		Admission:   admission,
 		RetryDelay:  time.Minute,
 	}
 	if err := processor.Process(context.Background(), retry); err == nil {
@@ -162,9 +138,6 @@ func TestProcessorRetriesBackendFailureAndTerminatesPoisonMessage(t *testing.T)
 	if retry.nacked != time.Minute {
 		t.Fatalf("retry delay = %s", retry.nacked)
 	}
-	if admission.released != 1 {
-		t.Fatalf("released slots = %d", admission.released)
-	}
 
 	poison := &fakeMessage{data: []byte("not-json")}
 	if err := processor.Process(context.Background(), poison); err == nil {
@@ -175,24 +148,6 @@ func TestProcessorRetriesBackendFailureAndTerminatesPoisonMessage(t *testing.T)
 	}
 }
 
-func TestProcessorLeavesAssignmentPendingWhenBackendPoolIsFull(t *testing.T) {
-	message := &fakeMessage{data: encodedAssignment(t)}
-	accepter := &fakeAccepter{accepted: true}
-	processor := Processor{
-		TrustDomain: "ddupan.top",
-		Accepter:    accepter,
-		Claims:      &fakeClaims{},
-		Admission:   &fakeAdmission{},
-		RetryDelay:  3 * time.Second,
-	}
-	if err := processor.Process(context.Background(), message); err != nil {
-		t.Fatal(err)
-	}
-	if message.acked != 0 || message.nacked != 3*time.Second {
-		t.Fatalf("message = %#v", message)
-	}
-}
-
 type fakeConsumerManager struct{ config jetstream.ConsumerConfig }
 
 func (m *fakeConsumerManager) CreateOrUpdateConsumer(_ context.Context, _ string, config jetstream.ConsumerConfig) (jetstream.Consumer, error) {

internal/backendpool/pool.go

@@ -1,50 +0,0 @@
-// Package backendpool manages runtime capacity independently for each executor backend.
-package backendpool
-
-import "sync"
-
-type Pool struct {
-	mu       sync.Mutex
-	capacity int
-	active   map[string]struct{}
-}
-
-func New(capacity int) *Pool {
-	return &Pool{capacity: capacity, active: make(map[string]struct{})}
-}
-
-// Acquire reserves a backend slot without blocking. Redelivery of the same
-// assignment is idempotent and succeeds even while the pool is full.
-func (p *Pool) Acquire(assignmentID string) bool {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	if _, exists := p.active[assignmentID]; exists {
-		return true
-	}
-	if assignmentID == "" || len(p.active) >= p.capacity {
-		return false
-	}
-	p.active[assignmentID] = struct{}{}
-	return true
-}
-
-func (p *Pool) Restore(assignmentID string) {
-	if assignmentID == "" {
-		return
-	}
-	p.mu.Lock()
-	p.active[assignmentID] = struct{}{}
-	p.mu.Unlock()
-}
-
-func (p *Pool) Release(assignmentID string) {
-	p.mu.Lock()
-	delete(p.active, assignmentID)
-	p.mu.Unlock()
-}
-
-func (p *Pool) Active() int {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	return len(p.active)
-}

internal/backendpool/pool_test.go

@@ -1,26 +0,0 @@
-package backendpool
-
-import "testing"
-
-func TestPoolSeparatesRuntimeCapacityFromDeliveries(t *testing.T) {
-	pool := New(2)
-	if !pool.Acquire("one") || !pool.Acquire("two") || pool.Acquire("three") {
-		t.Fatal("capacity was not enforced")
-	}
-	if !pool.Acquire("one") {
-		t.Fatal("redelivery must be idempotent")
-	}
-	pool.Release("one")
-	if !pool.Acquire("three") || pool.Active() != 2 {
-		t.Fatalf("active=%d", pool.Active())
-	}
-}
-
-func TestRestoreMayTemporarilyExceedReducedCapacity(t *testing.T) {
-	pool := New(1)
-	pool.Restore("one")
-	pool.Restore("two")
-	if pool.Active() != 2 || pool.Acquire("three") {
-		t.Fatalf("active=%d", pool.Active())
-	}
-}

internal/taskscheduler/poller.go

@@ -4,11 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sync/atomic"
 	"time"
 
-	"golang.org/x/sync/errgroup"
-
 	runnerv1 "gitea.dev/actionslib/runner/v1"
 )
 
@@ -22,12 +19,10 @@ type PollerConfig struct {
 	Labels       []string
 	EmptyBackoff time.Duration
 	ErrorBackoff time.Duration
-	Capacity     int
 }
 
-// Poller is the scheduler component. Each fetcher keeps its assigned task
+// Poller is the scheduler component. Once Gitea assigns a task, it never
-// until that assignment is durably dispatched; all fetchers share one runner
+// fetches another one until the current assignment is durably dispatched.
-// declaration and a monotonic tasks version.
 type Poller struct {
 	Client    PollClient
 	Scheduler *Scheduler
@@ -54,21 +49,9 @@ func (p Poller) Run(ctx context.Context) error {
 	if errorBackoff <= 0 {
 		errorBackoff = 5 * time.Second
 	}
-	capacity := p.Config.Capacity
+	var tasksVersion int64
-	if capacity < 1 {
-		capacity = 1
-	}
-	var tasksVersion atomic.Int64
-	group, groupContext := errgroup.WithContext(ctx)
-	for range capacity {
-		group.Go(func() error { return p.runFetcher(groupContext, &tasksVersion, emptyBackoff, errorBackoff) })
-	}
-	return group.Wait()
-}
-
-func (p Poller) runFetcher(ctx context.Context, tasksVersion *atomic.Int64, emptyBackoff, errorBackoff time.Duration) error {
 	for {
-		response, err := p.Client.FetchTask(ctx, tasksVersion.Load())
+		response, err := p.Client.FetchTask(ctx, tasksVersion)
 		if err != nil {
 			if ctx.Err() != nil {
 				return nil
@@ -86,7 +69,7 @@ func (p Poller) runFetcher(ctx context.Context, tasksVersion *atomic.Int64, empt
 			}
 			continue
 		}
-		storeMaximum(tasksVersion, response.GetTasksVersion())
+		tasksVersion = response.GetTasksVersion()
 		task := response.GetTask()
 		if task == nil {
 			if !wait(ctx, emptyBackoff) {
@@ -110,14 +93,6 @@ func (p Poller) runFetcher(ctx context.Context, tasksVersion *atomic.Int64, empt
 	}
 }
 
-func storeMaximum(value *atomic.Int64, candidate int64) {
-	for current := value.Load(); candidate > current; current = value.Load() {
-		if value.CompareAndSwap(current, candidate) {
-			return
-		}
-	}
-}
-
 func (p Poller) report(err error) {
 	if p.OnError != nil {
 		p.OnError(err)

internal/taskscheduler/poller_test.go

@@ -115,52 +115,3 @@ func TestPollerRetriesAssignedTaskBeforeFetchingAnother(t *testing.T) {
 		t.Fatalf("dispatches=%d fetches-before-dispatch=%d declares=%d", dispatcher.calls, dispatcher.fetchesAtSuccess, client.declared)
 	}
 }
-
-type blockingPollClient struct {
-	mu       sync.Mutex
-	declared int
-	started  chan struct{}
-}
-
-func (c *blockingPollClient) Declare(context.Context, string, []string) error {
-	c.mu.Lock()
-	c.declared++
-	c.mu.Unlock()
-	return nil
-}
-
-func (c *blockingPollClient) FetchTask(ctx context.Context, _ int64) (*runnerv1.FetchTaskResponse, error) {
-	c.started <- struct{}{}
-	<-ctx.Done()
-	return nil, ctx.Err()
-}
-
-func TestPollerStartsConfiguredNumberOfFetchersAfterOneDeclare(t *testing.T) {
-	client := &blockingPollClient{started: make(chan struct{}, 3)}
-	poller := Poller{
-		Client:    client,
-		Scheduler: &Scheduler{TrustDomain: "ddupan.top", Dispatcher: &retryDispatcher{done: make(chan struct{})}},
-		Config: PollerConfig{
-			Version: "dev", Labels: []string{"self-hosted:host", "pod:host", "vm:host"}, Capacity: 3,
-		},
-	}
-	ctx, cancel := context.WithCancel(context.Background())
-	finished := make(chan error, 1)
-	go func() { finished <- poller.Run(ctx) }()
-	for range 3 {
-		select {
-		case <-client.started:
-		case <-time.After(time.Second):
-			t.Fatal("configured fetchers did not start")
-		}
-	}
-	cancel()
-	if err := <-finished; err != nil {
-		t.Fatal(err)
-	}
-	client.mu.Lock()
-	defer client.mu.Unlock()
-	if client.declared != 1 {
-		t.Fatalf("declares = %d", client.declared)
-	}
-}