From 0d8b36380dabd3cac3f6b6db5211522bb5f66270 Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Sun, 20 Sep 2026 18:17:49 +0000 Subject: [PATCH] =?UTF-8?q?=E5=AE=9E=E7=8E=B0=E6=8C=81=E4=B9=85=E5=8C=96?= =?UTF-8?q?=20assignment=20handoff?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/runner-protocol-roadmap.md | 4 + go.mod | 7 +- go.sum | 14 ++- internal/assignmentqueue/jetstream.go | 96 +++++++++++++++ internal/assignmentqueue/jetstream_test.go | 130 +++++++++++++++++++++ internal/taskassignment/assignment.go | 125 ++++++++++++++++++++ internal/taskassignment/assignment_test.go | 75 ++++++++++++ internal/taskscheduler/scheduler.go | 18 +-- internal/taskworker/worker.go | 9 +- internal/taskworker/worker_test.go | 13 ++- 10 files changed, 464 insertions(+), 27 deletions(-) create mode 100644 internal/assignmentqueue/jetstream.go create mode 100644 internal/assignmentqueue/jetstream_test.go create mode 100644 internal/taskassignment/assignment.go create mode 100644 internal/taskassignment/assignment_test.go diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index 766642a..0a90051 100644 --- a/docs/runner-protocol-roadmap.md +++ b/docs/runner-protocol-roadmap.md @@ -37,6 +37,10 @@ dynamic-runner scheduler 当前 task 所需的短期 lease/capability。 - JetStream 只持久化和投递 assignment，不保存 executor 生命周期状态。Pod labels/annotations 与 OpenSandbox metadata 是后端运行状态的权威来源，Gitea 是 task 终态的权威来源。 +- assignment 使用版本化 envelope 保存完整 Gitea protobuf task，并从 workflow `runs-on` + 严格选择 pod 或 vm subject；消费者解码后重新派生 backend 与身份，拒绝被篡改的冗余字段。 +- JetStream 的 message ID 等于稳定 assignment ID `gitea-task-`，仅用于发布去重， + 不承担 executor 生命周期记录。 - worker 按稳定 assignment ID reconcile 后端资源，进程内只保留并发控制等可丢弃状态； 不新增数据库，也不依赖内存中的 runner-to-executor 映射。 - 结果处理顺序固定为回报 Gitea、清理后端、ACK assignment；重投时先查询 Gitea 终态， diff --git a/go.mod b/go.mod index 406657f..22a1391 100644 --- a/go.mod +++ b/go.mod @@ -5,11 +5,16 @@ go 1.27 require ( connectrpc.com/connect v1.20.0 gitea.dev/actionslib v1.0.0 + github.com/nats-io/nats.go v1.54.0 google.golang.org/protobuf v1.36.12 ) require ( + github.com/klauspost/compress v1.20.0 // indirect + github.com/nats-io/nkeys v0.4.16 // indirect + github.com/nats-io/nuid v1.0.1 // indirect github.com/sirupsen/logrus v1.10.2 // indirect go.yaml.in/yaml/v4 v4.0.0-rc.6 // indirect - golang.org/x/sys v0.46.0 // indirect + golang.org/x/crypto v0.57.0 // indirect + golang.org/x/sys v0.48.0 // indirect ) diff --git a/go.sum b/go.sum index f0c99dc..66a9465 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,14 @@ gitea.dev/actionslib v1.0.0 h1:l0oFJP+P4Ds1rlCI5zk618dYkuBc2mU7Gz5wPeG0lZY= gitea.dev/actionslib v1.0.0/go.mod h1:6O8YHkqVTKSR0LL2e5VhIDePYzGTZCbfmSVqJWEhk9g= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA= +github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI= +github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA= +github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k= +github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg= +github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/sirupsen/logrus v1.10.2 h1:G2SED73/qrAu6YwbdxOD6peLkCBI3z7L+ykJFTXJBBo= github.com/sirupsen/logrus v1.10.2/go.mod h1:SLEg8TqYulVKKfIGHldVp2K2aYz2DKSVBq4g/H5bR7Q= github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE= @@ -12,7 +20,9 @@ go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= go.yaml.in/yaml/v4 v4.0.0-rc.6 h1:1h7H1ohdUh93/FyE4YaDa1Zh64K6VVbjF4K6WUxMtH4= go.yaml.in/yaml/v4 v4.0.0-rc.6/go.mod h1:aZqd9kCMsGL7AuUv/m/PvWLdg5sjJsZ4oHDEnfPPfY0= -golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= -golang.org/x/sys v0.46.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= +golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= +golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= diff --git a/internal/assignmentqueue/jetstream.go b/internal/assignmentqueue/jetstream.go new file mode 100644 index 0000000..25e0c57 --- /dev/null +++ b/internal/assignmentqueue/jetstream.go @@ -0,0 +1,96 @@ +// Package assignmentqueue implements the durable assignment handoff with JetStream. +package assignmentqueue + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" +) + +type publishAPI interface { + PublishMsg(context.Context, *nats.Msg, ...jetstream.PublishOpt) (*jetstream.PubAck, error) +} + +// Publisher implements the scheduler dispatcher with one subject per backend. +type Publisher struct { + JetStream publishAPI + SubjectBase string +} + +func (p Publisher) Dispatch(ctx context.Context, assignment taskassignment.Assignment) error { + if p.JetStream == nil { + return errors.New("JetStream publisher is required") + } + body, err := taskassignment.Marshal(assignment) + if err != nil { + return err + } + base := strings.TrimSuffix(p.SubjectBase, ".") + if base == "" { + return errors.New("assignment subject base is required") + } + message := &nats.Msg{ + Subject: base + "." + string(assignment.Backend), + Header: nats.Header{jetstream.MsgIDHeader: []string{assignment.ID}}, + Data: body, + } + if _, err := p.JetStream.PublishMsg(ctx, message); err != nil { + return fmt.Errorf("publish assignment %s: %w", assignment.ID, err) + } + return nil +} + +type Handler interface { + Handle(context.Context, taskassignment.Assignment) (bool, error) +} + +// Message is the subset of jetstream.Msg needed by one reconciliation. +type Message interface { + Data() []byte + DoubleAck(context.Context) error + NakWithDelay(time.Duration) error + InProgress() error + TermWithReason(string) error +} + +// Processor maps one delivery to one idempotent worker reconciliation. +type Processor struct { + TrustDomain string + Handler Handler + RetryDelay time.Duration +} + +func (p Processor) Process(ctx context.Context, message Message) error { + if p.Handler == nil { + return errors.New("assignment handler is required") + } + assignment, err := taskassignment.Unmarshal(message.Data(), p.TrustDomain) + if err != nil { + return errors.Join(err, message.TermWithReason("invalid assignment")) + } + done, err := p.Handler.Handle(ctx, assignment) + if err != nil { + delay := p.RetryDelay + if delay <= 0 { + delay = 15 * time.Second + } + return errors.Join(err, message.NakWithDelay(delay)) + } + if done { + if err := message.DoubleAck(ctx); err != nil { + return fmt.Errorf("ack assignment %s: %w", assignment.ID, err) + } + return nil + } + if err := message.InProgress(); err != nil { + return fmt.Errorf("extend assignment %s acknowledgement: %w", assignment.ID, err) + } + return nil +} diff --git a/internal/assignmentqueue/jetstream_test.go b/internal/assignmentqueue/jetstream_test.go new file mode 100644 index 0000000..13619de --- /dev/null +++ b/internal/assignmentqueue/jetstream_test.go @@ -0,0 +1,130 @@ +package assignmentqueue + +import ( + "context" + "errors" + "testing" + "time" + + runnerv1 "gitea.dev/actionslib/runner/v1" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "google.golang.org/protobuf/types/known/structpb" + + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" +) + +func testAssignment(t *testing.T) taskassignment.Assignment { + t.Helper() + fields, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"}) + if err != nil { + t.Fatal(err) + } + assignment, err := taskassignment.New(&runnerv1.Task{ + Id: 42, + Context: fields, + WorkflowPayload: []byte("jobs:\n publish:\n runs-on: [self-hosted, pod]\n steps: []\n"), + }, "ddupan.top") + if err != nil { + t.Fatal(err) + } + return assignment +} + +type fakePublisher struct{ message *nats.Msg } + +func (p *fakePublisher) PublishMsg(_ context.Context, message *nats.Msg, _ ...jetstream.PublishOpt) (*jetstream.PubAck, error) { + p.message = message + return &jetstream.PubAck{}, nil +} + +func TestPublisherUsesBackendSubjectAndAssignmentDeduplication(t *testing.T) { + api := &fakePublisher{} + publisher := Publisher{JetStream: api, SubjectBase: "ci.assignment"} + if err := publisher.Dispatch(context.Background(), testAssignment(t)); err != nil { + t.Fatal(err) + } + if api.message.Subject != "ci.assignment.pod" { + t.Fatalf("subject = %q", api.message.Subject) + } + if api.message.Header.Get(jetstream.MsgIDHeader) != "gitea-task-42" { + t.Fatalf("message ID = %q", api.message.Header.Get(jetstream.MsgIDHeader)) + } +} + +type fakeHandler struct { + done bool + err error +} + +func (h fakeHandler) Handle(context.Context, taskassignment.Assignment) (bool, error) { + return h.done, h.err +} + +type fakeMessage struct { + data []byte + acked int + nacked time.Duration + inProgress int + terminated int +} + +func (m *fakeMessage) Data() []byte { return m.data } +func (m *fakeMessage) DoubleAck(context.Context) error { m.acked++; return nil } +func (m *fakeMessage) NakWithDelay(delay time.Duration) error { m.nacked = delay; return nil } +func (m *fakeMessage) InProgress() error { m.inProgress++; return nil } +func (m *fakeMessage) TermWithReason(string) error { m.terminated++; return nil } + +func encodedAssignment(t *testing.T) []byte { + t.Helper() + data, err := taskassignment.Marshal(testAssignment(t)) + if err != nil { + t.Fatal(err) + } + return data +} + +func TestProcessorAcknowledgesOnlyCompletedAssignment(t *testing.T) { + message := &fakeMessage{data: encodedAssignment(t)} + processor := Processor{TrustDomain: "ddupan.top", Handler: fakeHandler{done: true}} + if err := processor.Process(context.Background(), message); err != nil { + t.Fatal(err) + } + if message.acked != 1 || message.inProgress != 0 || message.nacked != 0 { + t.Fatalf("message = %#v", message) + } +} + +func TestProcessorKeepsRunningAssignmentPending(t *testing.T) { + message := &fakeMessage{data: encodedAssignment(t)} + processor := Processor{TrustDomain: "ddupan.top", Handler: fakeHandler{}} + if err := processor.Process(context.Background(), message); err != nil { + t.Fatal(err) + } + if message.acked != 0 || message.inProgress != 1 { + t.Fatalf("message = %#v", message) + } +} + +func TestProcessorRetriesBackendFailureAndTerminatesPoisonMessage(t *testing.T) { + retry := &fakeMessage{data: encodedAssignment(t)} + processor := Processor{ + TrustDomain: "ddupan.top", + Handler: fakeHandler{err: errors.New("backend unavailable")}, + RetryDelay: time.Minute, + } + if err := processor.Process(context.Background(), retry); err == nil { + t.Fatal("expected backend error") + } + if retry.nacked != time.Minute { + t.Fatalf("retry delay = %s", retry.nacked) + } + + poison := &fakeMessage{data: []byte("not-json")} + if err := processor.Process(context.Background(), poison); err == nil { + t.Fatal("expected decode error") + } + if poison.terminated != 1 || poison.nacked != 0 { + t.Fatalf("poison message = %#v", poison) + } +} diff --git a/internal/taskassignment/assignment.go b/internal/taskassignment/assignment.go new file mode 100644 index 0000000..6421774 --- /dev/null +++ b/internal/taskassignment/assignment.go @@ -0,0 +1,125 @@ +// Package taskassignment defines the durable handoff between the scheduler and workers. +package taskassignment + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "slices" + + "gitea.dev/actionslib/pkg/model" + runnerv1 "gitea.dev/actionslib/runner/v1" + "google.golang.org/protobuf/proto" + + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity" +) + +const wireVersion = 1 + +type Backend string + +const ( + BackendPod Backend = "pod" + BackendVM Backend = "vm" +) + +// Assignment is the only document persisted in the handoff queue. +type Assignment struct { + ID string + Backend Backend + Task *runnerv1.Task + Identity taskidentity.Identity +} + +type envelope struct { + Version int `json:"version"` + ID string `json:"id"` + Backend Backend `json:"backend"` + Task []byte `json:"task"` + Identity taskidentity.Identity `json:"identity"` +} + +// New derives all trusted assignment fields from the task fetched from Gitea. +func New(task *runnerv1.Task, trustDomain string) (Assignment, error) { + if task == nil || task.GetId() <= 0 { + return Assignment{}, errors.New("positive Gitea task ID is required") + } + identity, err := taskidentity.FromTask(task, trustDomain) + if err != nil { + return Assignment{}, err + } + backend, err := backendFromTask(task) + if err != nil { + return Assignment{}, err + } + return Assignment{ + ID: fmt.Sprintf("gitea-task-%d", task.GetId()), + Backend: backend, + Task: task, + Identity: identity, + }, nil +} + +func backendFromTask(task *runnerv1.Task) (Backend, error) { + workflow, err := model.ReadWorkflow(bytes.NewReader(task.GetWorkflowPayload())) + if err != nil { + return "", fmt.Errorf("parse task workflow for backend: %w", err) + } + jobIDs := workflow.GetJobIDs() + if len(jobIDs) != 1 || workflow.GetJob(jobIDs[0]) == nil { + return "", fmt.Errorf("task workflow must contain exactly one non-empty job") + } + labels := workflow.GetJob(jobIDs[0]).RunsOnLabels() + if !slices.Contains(labels, "self-hosted") { + return "", fmt.Errorf("task runs-on labels must include self-hosted: %v", labels) + } + hasPod := slices.Contains(labels, string(BackendPod)) + hasVM := slices.Contains(labels, string(BackendVM)) + if hasPod == hasVM { + return "", fmt.Errorf("task runs-on labels must select exactly one of pod or vm: %v", labels) + } + if hasPod { + return BackendPod, nil + } + return BackendVM, nil +} + +// Marshal encodes a versioned assignment. Protobuf preserves the exact Gitea task. +func Marshal(assignment Assignment) ([]byte, error) { + if assignment.Task == nil { + return nil, errors.New("assignment task is required") + } + task, err := proto.Marshal(assignment.Task) + if err != nil { + return nil, fmt.Errorf("marshal Gitea task: %w", err) + } + return json.Marshal(envelope{ + Version: wireVersion, + ID: assignment.ID, Backend: assignment.Backend, + Task: task, Identity: assignment.Identity, + }) +} + +// Unmarshal re-derives trusted fields instead of trusting duplicated queue metadata. +func Unmarshal(data []byte, trustDomain string) (Assignment, error) { + var wire envelope + if err := json.Unmarshal(data, &wire); err != nil { + return Assignment{}, fmt.Errorf("decode assignment: %w", err) + } + if wire.Version != wireVersion { + return Assignment{}, fmt.Errorf("unsupported assignment version %d", wire.Version) + } + task := new(runnerv1.Task) + if err := proto.Unmarshal(wire.Task, task); err != nil { + return Assignment{}, fmt.Errorf("unmarshal Gitea task: %w", err) + } + canonical, err := New(task, trustDomain) + if err != nil { + return Assignment{}, err + } + if wire.ID != canonical.ID || wire.Backend != canonical.Backend || wire.Identity != canonical.Identity { + return Assignment{}, errors.New("assignment metadata does not match its Gitea task") + } + return canonical, nil +} diff --git a/internal/taskassignment/assignment_test.go b/internal/taskassignment/assignment_test.go new file mode 100644 index 0000000..0646e30 --- /dev/null +++ b/internal/taskassignment/assignment_test.go @@ -0,0 +1,75 @@ +package taskassignment + +import ( + "bytes" + "testing" + + runnerv1 "gitea.dev/actionslib/runner/v1" + "google.golang.org/protobuf/types/known/structpb" +) + +func task(t *testing.T, labels string) *runnerv1.Task { + t.Helper() + context, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"}) + if err != nil { + t.Fatal(err) + } + return &runnerv1.Task{ + Id: 42, + Context: context, + WorkflowPayload: []byte("jobs:\n publish:\n runs-on: " + labels + "\n steps: []\n"), + } +} + +func TestNewSelectsBackendFromRunsOn(t *testing.T) { + for _, test := range []struct { + labels string + backend Backend + }{ + {"[self-hosted, pod]", BackendPod}, + {"[self-hosted, vm]", BackendVM}, + } { + assignment, err := New(task(t, test.labels), "ddupan.top") + if err != nil { + t.Fatal(err) + } + if assignment.Backend != test.backend || assignment.ID != "gitea-task-42" { + t.Fatalf("assignment = %#v", assignment) + } + } +} + +func TestNewRejectsAmbiguousBackend(t *testing.T) { + for _, labels := range []string{ + "[self-hosted]", + "[self-hosted, pod, vm]", + "[pod]", + } { + if _, err := New(task(t, labels), "ddupan.top"); err == nil { + t.Fatalf("expected labels %s to fail", labels) + } + } +} + +func TestAssignmentWireRoundTripAndValidation(t *testing.T) { + want, err := New(task(t, "[self-hosted, pod]"), "ddupan.top") + if err != nil { + t.Fatal(err) + } + data, err := Marshal(want) + if err != nil { + t.Fatal(err) + } + got, err := Unmarshal(data, "ddupan.top") + if err != nil { + t.Fatal(err) + } + if got.ID != want.ID || got.Backend != want.Backend || got.Identity != want.Identity || !bytes.Equal(got.Task.WorkflowPayload, want.Task.WorkflowPayload) { + t.Fatalf("round trip = %#v, want %#v", got, want) + } + + tampered := bytes.Replace(data, []byte(`"backend":"pod"`), []byte(`"backend":"vm"`), 1) + if _, err := Unmarshal(tampered, "ddupan.top"); err == nil { + t.Fatal("expected tampered backend to fail") + } +} diff --git a/internal/taskscheduler/scheduler.go b/internal/taskscheduler/scheduler.go index 3f05753..feae766 100644 --- a/internal/taskscheduler/scheduler.go +++ b/internal/taskscheduler/scheduler.go @@ -5,19 +5,13 @@ package taskscheduler import ( "context" "errors" - "fmt" runnerv1 "gitea.dev/actionslib/runner/v1" - "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" ) -// Assignment is the immutable input handed to a Pod or VM provisioner. -type Assignment struct { - ID string - Task *runnerv1.Task - Identity taskidentity.Identity -} +type Assignment = taskassignment.Assignment // Dispatcher creates exactly one executor for an already assigned Gitea task. // It must not register another runner or ask Gitea for a task. @@ -36,13 +30,9 @@ func (s *Scheduler) Run(ctx context.Context, task *runnerv1.Task) error { if s.Dispatcher == nil { return errors.New("executor dispatcher is required") } - identity, err := taskidentity.FromTask(task, s.TrustDomain) + assignment, err := taskassignment.New(task, s.TrustDomain) if err != nil { return err } - return s.Dispatcher.Dispatch(ctx, Assignment{ - ID: fmt.Sprintf("gitea-task-%d", task.GetId()), - Task: task, - Identity: identity, - }) + return s.Dispatcher.Dispatch(ctx, assignment) } diff --git a/internal/taskworker/worker.go b/internal/taskworker/worker.go index ac3fab0..1e71ab2 100644 --- a/internal/taskworker/worker.go +++ b/internal/taskworker/worker.go @@ -6,8 +6,8 @@ import ( "errors" "strconv" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" "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. @@ -38,7 +38,7 @@ type Metadata struct { // Every method must be idempotent. type Backend interface { Find(context.Context, string) (*Executor, error) - Create(context.Context, taskscheduler.Assignment, Metadata) (*Executor, error) + Create(context.Context, taskassignment.Assignment, Metadata) (*Executor, error) BindIdentity(context.Context, *Executor, taskidentity.Identity) error Delete(context.Context, *Executor) error } @@ -59,7 +59,7 @@ type Worker struct { // 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) { +func (w Worker) Handle(ctx context.Context, assignment taskassignment.Assignment) (done bool, err error) { if w.Backend == nil || w.Tasks == nil { return false, errors.New("backend and Gitea task state are required") } @@ -114,11 +114,12 @@ func (w Worker) Handle(ctx context.Context, assignment taskscheduler.Assignment) } // BackendMetadata is the shared metadata contract for Pods and OpenSandbox. -func BackendMetadata(assignment taskscheduler.Assignment) Metadata { +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, diff --git a/internal/taskworker/worker_test.go b/internal/taskworker/worker_test.go index c115ca0..17b5494 100644 --- a/internal/taskworker/worker_test.go +++ b/internal/taskworker/worker_test.go @@ -6,8 +6,8 @@ import ( runnerv1 "gitea.dev/actionslib/runner/v1" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity" - "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskscheduler" ) type fakeBackend struct { @@ -18,7 +18,7 @@ type fakeBackend struct { } func (b *fakeBackend) Find(context.Context, string) (*Executor, error) { return b.executor, nil } -func (b *fakeBackend) Create(_ context.Context, _ taskscheduler.Assignment, _ Metadata) (*Executor, error) { +func (b *fakeBackend) Create(_ context.Context, _ taskassignment.Assignment, _ Metadata) (*Executor, error) { b.created++ b.executor = &Executor{Name: "executor", IdentityTarget: "pod-uid", Phase: PhaseRunning} return b.executor, nil @@ -45,10 +45,11 @@ func (t *fakeTasks) Report(_ context.Context, _ int64, phase Phase) error { return nil } -func assignment() taskscheduler.Assignment { - return taskscheduler.Assignment{ - ID: "gitea-task-42", - Task: &runnerv1.Task{Id: 42}, +func assignment() taskassignment.Assignment { + return taskassignment.Assignment{ + ID: "gitea-task-42", + Backend: taskassignment.BackendPod, + Task: &runnerv1.Task{Id: 42}, Identity: taskidentity.Identity{ Repository: "owner/repo", Task: "publish",