From cbbe14d6b5a9267a27c42b0e077269d56f02a14b Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Sun, 20 Sep 2026 20:53:58 +0000 Subject: [PATCH] fix: gate scheduler task concurrency --- cmd/gitea-dynamic-runner/controller.go | 8 +++++-- docs/runner-protocol-roadmap.md | 3 +++ internal/runnerfacade/facade.go | 7 +++++- internal/runnerfacade/facade_test.go | 23 +++++++++++++++++++ internal/taskscheduler/gate.go | 31 ++++++++++++++++++++++++++ internal/taskscheduler/gate_test.go | 29 ++++++++++++++++++++++++ internal/taskscheduler/poller.go | 13 +++++++++++ 7 files changed, 111 insertions(+), 3 deletions(-) create mode 100644 internal/taskscheduler/gate.go create mode 100644 internal/taskscheduler/gate_test.go diff --git a/cmd/gitea-dynamic-runner/controller.go b/cmd/gitea-dynamic-runner/controller.go index 2e2b1ee..f7e9d01 100644 --- a/cmd/gitea-dynamic-runner/controller.go +++ b/cmd/gitea-dynamic-runner/controller.go @@ -85,8 +85,12 @@ func runController(ctx context.Context) error { return err } registry := runnerfacade.NewRegistry() + gate := taskscheduler.NewSingleFlightGate() giteaClient := giteaactions.NewClient(giteaactions.DefaultHTTPClient(), config.GiteaURL, config.GiteaUUID, config.GiteaToken) - facade := &runnerfacade.Facade{Registry: registry, Capabilities: capabilities, Upstream: giteaClient} + facade := &runnerfacade.Facade{ + Registry: registry, Capabilities: capabilities, Upstream: giteaClient, + OnTerminal: func(taskassignment.Assignment) { gate.Release() }, + } bootstrap := runnerbootstrap.Bootstrap{ Capabilities: capabilities, FacadeURL: config.FacadeURL, FacadeSPIFFEID: config.FacadeSPIFFEID, } @@ -99,7 +103,7 @@ func runController(ctx context.Context) error { labels = append(labels, string(taskassignment.BackendVM)) } poller := taskscheduler.Poller{ - Client: giteaClient, + Client: giteaClient, Gate: gate, Scheduler: &taskscheduler.Scheduler{TrustDomain: config.TrustDomain, Dispatcher: assignmentqueue.Publisher{ JetStream: producerJS, SubjectBase: config.SubjectBase, }}, diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index 8dea173..f8c0ed4 100644 --- a/docs/runner-protocol-roadmap.md +++ b/docs/runner-protocol-roadmap.md @@ -79,6 +79,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 实际运行资源为准。 - 两种 backend 都注入同一份 runner bootstrap 环境；Pod 仍由 homelab Kubernetes 原生 创建，只有 VM 经 OpenSandbox 创建，bootstrap 机制不改变 backend 边界。 diff --git a/internal/runnerfacade/facade.go b/internal/runnerfacade/facade.go index c6cc84e..b713487 100644 --- a/internal/runnerfacade/facade.go +++ b/internal/runnerfacade/facade.go @@ -57,6 +57,7 @@ type Facade struct { Registry *Registry Capabilities Capabilities Upstream Upstream + OnTerminal func(taskassignment.Assignment) } func (f *Facade) Handler() (string, http.Handler) { @@ -98,7 +99,11 @@ func (f *Facade) UpdateTask(ctx context.Context, request *connect.Request[runner if request.Msg.GetState().GetId() != assignment.Task.GetId() { return nil, connect.NewError(connect.CodePermissionDenied, errors.New("task update does not match assignment")) } - return f.Upstream.UpdateTask(ctx, connect.NewRequest(request.Msg)) + response, err := f.Upstream.UpdateTask(ctx, connect.NewRequest(request.Msg)) + if err == nil && request.Msg.GetState().GetResult() != runnerv1.Result_RESULT_UNSPECIFIED && f.OnTerminal != nil { + f.OnTerminal(assignment) + } + return response, err } func (f *Facade) UpdateLog(ctx context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) { diff --git a/internal/runnerfacade/facade_test.go b/internal/runnerfacade/facade_test.go index e8eaa2f..dca6821 100644 --- a/internal/runnerfacade/facade_test.go +++ b/internal/runnerfacade/facade_test.go @@ -163,6 +163,29 @@ func TestFacadeForwardsOnlyMatchingTaskAndLogUpdates(t *testing.T) { } } +func TestFacadeSignalsTerminalTaskAfterUpstreamAcceptsIt(t *testing.T) { + facade, assignment, token := testFacade(t) + ctx := WithSPIFFEID(context.Background(), assignment.Identity.SPIFFEID) + if _, err := facade.FetchTask(ctx, authenticatedRequest(&runnerv1.FetchTaskRequest{}, assignment.ID, token)); err != nil { + t.Fatal(err) + } + completed := 0 + facade.OnTerminal = func(got taskassignment.Assignment) { + if got.ID != assignment.ID { + t.Fatalf("terminal assignment = %s", got.ID) + } + completed++ + } + if _, err := facade.UpdateTask(ctx, authenticatedRequest(&runnerv1.UpdateTaskRequest{ + State: &runnerv1.TaskState{Id: 42, Result: runnerv1.Result_RESULT_SUCCESS}, + }, assignment.ID, token)); err != nil { + t.Fatal(err) + } + if completed != 1 { + t.Fatalf("terminal notifications = %d", completed) + } +} + func TestCapabilitiesAreDeterministicAndAssignmentScoped(t *testing.T) { capabilities, err := NewCapabilities([]byte("0123456789abcdef0123456789abcdef")) if err != nil { diff --git a/internal/taskscheduler/gate.go b/internal/taskscheduler/gate.go new file mode 100644 index 0000000..b88c2d7 --- /dev/null +++ b/internal/taskscheduler/gate.go @@ -0,0 +1,31 @@ +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: + } +} diff --git a/internal/taskscheduler/gate_test.go b/internal/taskscheduler/gate_test.go new file mode 100644 index 0000000..25815d0 --- /dev/null +++ b/internal/taskscheduler/gate_test.go @@ -0,0 +1,29 @@ +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") + } +} diff --git a/internal/taskscheduler/poller.go b/internal/taskscheduler/poller.go index b03faaf..6447e73 100644 --- a/internal/taskscheduler/poller.go +++ b/internal/taskscheduler/poller.go @@ -26,6 +26,7 @@ type PollerConfig struct { type Poller struct { Client PollClient Scheduler *Scheduler + Gate *SingleFlightGate Config PollerConfig OnError func(error) } @@ -50,7 +51,14 @@ 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 { @@ -72,6 +80,10 @@ 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 } @@ -79,6 +91,7 @@ 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 {