diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index 3e7ebba..90b3ffb 100644 --- a/docs/runner-protocol-roadmap.md +++ b/docs/runner-protocol-roadmap.md @@ -30,6 +30,8 @@ controller 使用单一 Go 二进制；默认在同一进程启用 `scheduler` - 对 workflow 的接口保持 `[self-hosted, pod]` 和 `[self-hosted, vm]` 不变。 - scheduler 在没有对应 backend 容量时不领取 task，避免本地形成不可控积压。 +- scheduler Declare 后使用 RunnerService 长轮询；一旦 FetchTask 返回已分配 task，在 + JetStream publish 成功前只重试该 assignment，不领取下一项。 - 每个 executor 只执行一个 task，完成后销毁。 - SPIFFE 身份从实际领取的 task 的 repository 和 workflow job key 派生，不需要 queued 与 in-progress webhook 的二阶段关联。 @@ -63,7 +65,8 @@ controller 使用单一 Go 二进制；默认在同一进程启用 `scheduler` 1. 固定当前 Gitea 版本所使用的 RunnerService protobuf 与 act_runner 版本，记录兼容 范围并建立协议契约测试。 2. 实现只注册、Declare labels 和容量感知 FetchTask 的 scheduler spike，暂不执行 - task。 + task。首次集成必须验证 FetchTask 后、JetStream publish 前进程崩溃时 Gitea 对同一 + runner 的 task 恢复语义；该窗口未验证前不能声称 scheduler 可无损恢复。 3. 从 act_runner 提取或复用 task 执行与日志上报能力，定义 scheduler 到 executor 的 单任务协议。 4. 首先接入 Pod executor，验证成功、失败、取消、超时和 scheduler 重启。 diff --git a/internal/taskscheduler/poller.go b/internal/taskscheduler/poller.go new file mode 100644 index 0000000..b03faaf --- /dev/null +++ b/internal/taskscheduler/poller.go @@ -0,0 +1,111 @@ +package taskscheduler + +import ( + "context" + "errors" + "fmt" + "time" + + runnerv1 "gitea.dev/actionslib/runner/v1" +) + +type PollClient interface { + Declare(context.Context, string, []string) error + FetchTask(context.Context, int64) (*runnerv1.FetchTaskResponse, error) +} + +type PollerConfig struct { + Version string + Labels []string + EmptyBackoff time.Duration + ErrorBackoff time.Duration +} + +// Poller is the scheduler component. Once Gitea assigns a task, it never +// fetches another one until the current assignment is durably dispatched. +type Poller struct { + Client PollClient + Scheduler *Scheduler + Config PollerConfig + OnError func(error) +} + +func (p Poller) Run(ctx context.Context) error { + if p.Client == nil || p.Scheduler == nil { + return errors.New("Gitea poll client and task scheduler are required") + } + if p.Config.Version == "" || len(p.Config.Labels) == 0 { + return errors.New("runner version and labels are required") + } + if err := p.Client.Declare(ctx, p.Config.Version, p.Config.Labels); err != nil { + return fmt.Errorf("declare scheduler labels: %w", err) + } + + emptyBackoff := p.Config.EmptyBackoff + if emptyBackoff <= 0 { + emptyBackoff = time.Second + } + errorBackoff := p.Config.ErrorBackoff + if errorBackoff <= 0 { + errorBackoff = 5 * time.Second + } + var tasksVersion int64 + for { + response, err := p.Client.FetchTask(ctx, tasksVersion) + if err != nil { + if ctx.Err() != nil { + return nil + } + p.report(fmt.Errorf("fetch Gitea task: %w", err)) + if !wait(ctx, errorBackoff) { + return nil + } + continue + } + if response == nil { + p.report(errors.New("fetch Gitea task returned an empty response")) + if !wait(ctx, errorBackoff) { + return nil + } + continue + } + tasksVersion = response.GetTasksVersion() + task := response.GetTask() + if task == nil { + if !wait(ctx, emptyBackoff) { + return nil + } + continue + } + for { + if err := p.Scheduler.Run(ctx, task); err == nil { + break + } else { + if ctx.Err() != nil { + return nil + } + p.report(fmt.Errorf("dispatch Gitea task %d: %w", task.GetId(), err)) + } + if !wait(ctx, errorBackoff) { + return nil + } + } + } +} + +func (p Poller) report(err error) { + if p.OnError != nil { + p.OnError(err) + } +} + +func wait(ctx context.Context, duration time.Duration) bool { + timer := time.NewTimer(duration) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} diff --git a/internal/taskscheduler/poller_test.go b/internal/taskscheduler/poller_test.go new file mode 100644 index 0000000..dbc0de1 --- /dev/null +++ b/internal/taskscheduler/poller_test.go @@ -0,0 +1,117 @@ +package taskscheduler + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + runnerv1 "gitea.dev/actionslib/runner/v1" + "google.golang.org/protobuf/types/known/structpb" +) + +type fakePollClient struct { + mu sync.Mutex + declared int + fetches int + responses []*runnerv1.FetchTaskResponse +} + +func (c *fakePollClient) Declare(context.Context, string, []string) error { + c.mu.Lock() + defer c.mu.Unlock() + c.declared++ + return nil +} + +func (c *fakePollClient) FetchTask(ctx context.Context, _ int64) (*runnerv1.FetchTaskResponse, error) { + c.mu.Lock() + c.fetches++ + if len(c.responses) > 0 { + response := c.responses[0] + c.responses = c.responses[1:] + c.mu.Unlock() + return response, nil + } + c.mu.Unlock() + <-ctx.Done() + return nil, ctx.Err() +} + +type retryDispatcher struct { + mu sync.Mutex + calls int + failures int + done chan struct{} + fetches func() int + fetchesAtSuccess int +} + +func (d *retryDispatcher) Dispatch(context.Context, Assignment) error { + d.mu.Lock() + defer d.mu.Unlock() + d.calls++ + if d.failures > 0 { + d.failures-- + return errors.New("JetStream unavailable") + } + if d.fetches != nil { + d.fetchesAtSuccess = d.fetches() + } + select { + case <-d.done: + default: + close(d.done) + } + return nil +} + +func pollTask(t *testing.T) *runnerv1.Task { + t.Helper() + fields, err := structpb.NewStruct(map[string]any{"repository": "owner/repo"}) + if err != nil { + t.Fatal(err) + } + return &runnerv1.Task{ + Id: 42, Context: fields, + WorkflowPayload: []byte("jobs:\n test:\n runs-on: [self-hosted, pod]\n steps: []\n"), + } +} + +func TestPollerRetriesAssignedTaskBeforeFetchingAnother(t *testing.T) { + client := &fakePollClient{responses: []*runnerv1.FetchTaskResponse{{Task: pollTask(t), TasksVersion: 7}}} + dispatcher := &retryDispatcher{failures: 2, done: make(chan struct{})} + dispatcher.fetches = func() int { + client.mu.Lock() + defer client.mu.Unlock() + return client.fetches + } + poller := Poller{ + Client: client, + Scheduler: &Scheduler{TrustDomain: "ddupan.top", Dispatcher: dispatcher}, + Config: PollerConfig{ + Version: "dev", Labels: []string{"self-hosted:host", "pod:host", "vm:host"}, + EmptyBackoff: time.Millisecond, ErrorBackoff: time.Millisecond, + }, + } + ctx, cancel := context.WithCancel(context.Background()) + finished := make(chan error, 1) + go func() { finished <- poller.Run(ctx) }() + select { + case <-dispatcher.done: + case <-time.After(time.Second): + t.Fatal("assignment was not dispatched") + } + cancel() + if err := <-finished; err != nil { + t.Fatal(err) + } + dispatcher.mu.Lock() + defer dispatcher.mu.Unlock() + client.mu.Lock() + defer client.mu.Unlock() + if dispatcher.calls != 3 || dispatcher.fetchesAtSuccess != 1 || client.declared != 1 { + t.Fatalf("dispatches=%d fetches-before-dispatch=%d declares=%d", dispatcher.calls, dispatcher.fetchesAtSuccess, client.declared) + } +}