From 22dc3fef63e764789bdea1ba37f82433752e7e0b Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Sun, 20 Sep 2026 18:11:57 +0000 Subject: [PATCH] =?UTF-8?q?=E5=BB=BA=E7=AB=8B=E6=97=A0=E7=8A=B6=E6=80=81?= =?UTF-8?q?=20worker=20reconcile=20=E8=BE=B9=E7=95=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/runner-protocol-roadmap.md | 8 +- internal/taskscheduler/scheduler.go | 8 +- internal/taskscheduler/scheduler_test.go | 3 + internal/taskworker/worker.go | 129 +++++++++++++++++++++++ internal/taskworker/worker_test.go | 102 ++++++++++++++++++ 5 files changed, 247 insertions(+), 3 deletions(-) create mode 100644 internal/taskworker/worker.go create mode 100644 internal/taskworker/worker_test.go diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index 80d98f3..766642a 100644 --- a/docs/runner-protocol-roadmap.md +++ b/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， - 但不是 Gitea 任务事实来源。 +- JetStream 只持久化和投递 assignment，不保存 executor 生命周期状态。Pod labels/annotations + 与 OpenSandbox metadata 是后端运行状态的权威来源，Gitea 是 task 终态的权威来源。 +- worker 按稳定 assignment ID reconcile 后端资源，进程内只保留并发控制等可丢弃状态； + 不新增数据库，也不依赖内存中的 runner-to-executor 映射。 +- 结果处理顺序固定为回报 Gitea、清理后端、ACK assignment；重投时先查询 Gitea 终态， + 从而关闭后端已删除但消息尚未 ACK 的崩溃窗口。 - Pod 与 VM 共享 task/executor 协议，只有环境创建和销毁实现不同。 ## 实现顺序 diff --git a/internal/taskscheduler/scheduler.go b/internal/taskscheduler/scheduler.go index c879400..3f05753 100644 --- a/internal/taskscheduler/scheduler.go +++ b/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, + }) } diff --git a/internal/taskscheduler/scheduler_test.go b/internal/taskscheduler/scheduler_test.go index 1e394f9..c19cbb3 100644 --- a/internal/taskscheduler/scheduler_test.go +++ b/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) diff --git a/internal/taskworker/worker.go b/internal/taskworker/worker.go new file mode 100644 index 0000000..ac3fab0 --- /dev/null +++ b/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, + }, + } +} diff --git a/internal/taskworker/worker_test.go b/internal/taskworker/worker_test.go new file mode 100644 index 0000000..c115ca0 --- /dev/null +++ b/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) + } +}