建立无状态 worker reconcile 边界

docs/runner-protocol-roadmap.md

@@ -35,8 +35,12 @@ dynamic-runner scheduler
   OpenBao 等资源服务继续是唯一授权决策点。
 - scheduler 的 runner registration credential 不进入 executor；executor 只得到执行
   当前 task 所需的短期 lease/capability。
-- task ACK、心跳和结果必须能够跨 scheduler 重启恢复；NATS 可以继续作为内部 handoff，
+- JetStream 只持久化和投递 assignment，不保存 executor 生命周期状态。Pod labels/annotations
-  但不是 Gitea 任务事实来源。
+  与 OpenSandbox metadata 是后端运行状态的权威来源，Gitea 是 task 终态的权威来源。
+- worker 按稳定 assignment ID reconcile 后端资源，进程内只保留并发控制等可丢弃状态；
+  不新增数据库，也不依赖内存中的 runner-to-executor 映射。
+- 结果处理顺序固定为回报 Gitea、清理后端、ACK assignment；重投时先查询 Gitea 终态，
+  从而关闭后端已删除但消息尚未 ACK 的崩溃窗口。
 - Pod 与 VM 共享 task/executor 协议，只有环境创建和销毁实现不同。
 
 ## 实现顺序

internal/taskscheduler/scheduler.go

@@ -5,6 +5,7 @@ package taskscheduler
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	runnerv1 "gitea.dev/actionslib/runner/v1"
 
@@ -13,6 +14,7 @@ import (
 
 // Assignment is the immutable input handed to a Pod or VM provisioner.
 type Assignment struct {
+	ID       string
 	Task     *runnerv1.Task
 	Identity taskidentity.Identity
 }
@@ -38,5 +40,9 @@ func (s *Scheduler) Run(ctx context.Context, task *runnerv1.Task) error {
 	if err != nil {
 		return err
 	}
-	return s.Dispatcher.Dispatch(ctx, Assignment{Task: task, Identity: identity})
+	return s.Dispatcher.Dispatch(ctx, Assignment{
+		ID:       fmt.Sprintf("gitea-task-%d", task.GetId()),
+		Task:     task,
+		Identity: identity,
+	})
 }

internal/taskscheduler/scheduler_test.go

@@ -39,6 +39,9 @@ func TestRunDerivesIdentityBeforeDispatch(t *testing.T) {
 	if dispatcher.assignment.Task != task {
 		t.Fatal("dispatcher did not receive the fetched task")
 	}
+	if dispatcher.assignment.ID != "gitea-task-42" {
+		t.Fatalf("assignment ID = %q", dispatcher.assignment.ID)
+	}
 	want := "spiffe://ddupan.top/ci/panxiao81/gitea-dynamic-runner/test"
 	if dispatcher.assignment.Identity.SPIFFEID != want {
 		t.Fatalf("SPIFFE ID = %q, want %q", dispatcher.assignment.Identity.SPIFFEID, want)

internal/taskworker/worker.go

@@ -0,0 +1,129 @@
+// Package taskworker reconciles assigned Gitea tasks against an executor backend.
+package taskworker
+
+import (
+	"context"
+	"errors"
+	"strconv"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskscheduler"
+)
+
+// 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
+}
+
+// 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, taskscheduler.Assignment, Metadata) (*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
+}
+
+// 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 taskscheduler.Assignment) (done bool, err error) {
+	if w.Backend == nil || w.Tasks == nil {
+		return false, errors.New("backend and Gitea task state 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 {
+		executor, err = w.Backend.Create(ctx, assignment, BackendMetadata(assignment))
+		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
+	}
+}
+
+// BackendMetadata is the shared metadata contract for Pods and OpenSandbox.
+func BackendMetadata(assignment taskscheduler.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),
+		},
+		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,
+		},
+	}
+}

internal/taskworker/worker_test.go

@@ -0,0 +1,102 @@
+package taskworker
+
+import (
+	"context"
+	"testing"
+
+	runnerv1 "gitea.dev/actionslib/runner/v1"
+
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity"
+	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskscheduler"
+)
+
+type fakeBackend struct {
+	executor *Executor
+	created  int
+	bound    int
+	deleted  int
+}
+
+func (b *fakeBackend) Find(context.Context, string) (*Executor, error) { return b.executor, nil }
+func (b *fakeBackend) Create(_ context.Context, _ taskscheduler.Assignment, _ Metadata) (*Executor, error) {
+	b.created++
+	b.executor = &Executor{Name: "executor", IdentityTarget: "pod-uid", Phase: PhaseRunning}
+	return b.executor, nil
+}
+func (b *fakeBackend) BindIdentity(context.Context, *Executor, taskidentity.Identity) error {
+	b.bound++
+	return nil
+}
+func (b *fakeBackend) Delete(context.Context, *Executor) error {
+	b.deleted++
+	b.executor = nil
+	return nil
+}
+
+type fakeTasks struct {
+	terminal bool
+	reported []Phase
+}
+
+func (t *fakeTasks) Terminal(context.Context, int64) (bool, error) { return t.terminal, nil }
+func (t *fakeTasks) Report(_ context.Context, _ int64, phase Phase) error {
+	t.reported = append(t.reported, phase)
+	t.terminal = true
+	return nil
+}
+
+func assignment() taskscheduler.Assignment {
+	return taskscheduler.Assignment{
+		ID:   "gitea-task-42",
+		Task: &runnerv1.Task{Id: 42},
+		Identity: taskidentity.Identity{
+			Repository: "owner/repo",
+			Task:       "publish",
+			SPIFFEID:   "spiffe://ddupan.top/ci/owner/repo/publish",
+		},
+	}
+}
+
+func TestHandleRecoversExistingExecutorWithoutCreatingAnother(t *testing.T) {
+	backend := &fakeBackend{executor: &Executor{Name: "existing", IdentityTarget: "uid", Phase: PhaseRunning}}
+	worker := Worker{Backend: backend, Tasks: &fakeTasks{}}
+
+	done, err := worker.Handle(context.Background(), assignment())
+	if err != nil || done {
+		t.Fatalf("Handle() = (%v, %v), want pending", done, err)
+	}
+	if backend.created != 0 || backend.bound != 1 {
+		t.Fatalf("created=%d bound=%d", backend.created, backend.bound)
+	}
+}
+
+func TestHandleReportsBeforeCleanupAndBecomesRecoverable(t *testing.T) {
+	backend := &fakeBackend{executor: &Executor{Name: "finished", IdentityTarget: "uid", Phase: PhaseSucceeded}}
+	tasks := &fakeTasks{}
+	worker := Worker{Backend: backend, Tasks: tasks}
+
+	done, err := worker.Handle(context.Background(), assignment())
+	if err != nil || !done {
+		t.Fatalf("Handle() = (%v, %v), want done", done, err)
+	}
+	if len(tasks.reported) != 1 || backend.deleted != 1 {
+		t.Fatalf("reported=%v deleted=%d", tasks.reported, backend.deleted)
+	}
+
+	// Simulate redelivery after deletion but before the queue ACK. Gitea's
+	// terminal state prevents a duplicate executor from being created.
+	done, err = worker.Handle(context.Background(), assignment())
+	if err != nil || !done || backend.created != 0 {
+		t.Fatalf("recovery = (%v, %v), created=%d", done, err, backend.created)
+	}
+}
+
+func TestBackendMetadataContainsRecoveryKeys(t *testing.T) {
+	metadata := BackendMetadata(assignment())
+	if metadata.Labels["ci.ddupan.top/assignment-id"] != "gitea-task-42" {
+		t.Fatalf("labels = %#v", metadata.Labels)
+	}
+	if metadata.Annotations["ci.ddupan.top/spiffe-id"] != "spiffe://ddupan.top/ci/owner/repo/publish" {
+		t.Fatalf("annotations = %#v", metadata.Annotations)
+	}
+}