From 59d8bbcde19d2eef0151aa43f101458ab415fa34 Mon Sep 17 00:00:00 2001 From: Noel Date: Fri, 7 Aug 2026 20:05:12 +0200 Subject: [PATCH] fix(actions): refresh watch client before polls --- cmd/actions/runs/watch.go | 4 +- modules/task/actions_run_watch.go | 81 +++++++++++++++++++++----- modules/task/actions_run_watch_test.go | 60 +++++++++++++++++++ 3 files changed, 131 insertions(+), 14 deletions(-) diff --git a/cmd/actions/runs/watch.go b/cmd/actions/runs/watch.go index d14f04e8..29e85039 100644 --- a/cmd/actions/runs/watch.go +++ b/cmd/actions/runs/watch.go @@ -74,7 +74,9 @@ func runRunsWatch(ctx stdctx.Context, cmd *cli.Command) error { if err := c.Ensure(context.CtxRequirement{RemoteRepo: true}); err != nil { return cli.Exit(err, task.ActionRunWatchExitError) } - result, err := task.WatchActionRuns(ctx, c.Login.Client().Actions, c.Owner, c.Repo, task.ActionRunWatchOptions{ + result, err := task.WatchActionRunsWithClientFactory(ctx, func() (task.ActionRunWatchClient, error) { + return c.Login.Client().Actions, nil + }, c.Owner, c.Repo, task.ActionRunWatchOptions{ RunIDs: runIDs, ExpectedHead: cmd.String("expect-head"), Interval: cmd.Duration("interval"), diff --git a/modules/task/actions_run_watch.go b/modules/task/actions_run_watch.go index e5ba9c9d..eb76154d 100644 --- a/modules/task/actions_run_watch.go +++ b/modules/task/actions_run_watch.go @@ -35,6 +35,11 @@ type ActionRunWatchClient interface { ListRepoJobsByRun(stdctx.Context, string, string, int64, gitea.ListRepoActionsJobsOptions) (*gitea.ActionWorkflowJobsResponse, *gitea.Response, error) } +// ActionRunWatchClientFactory creates an authenticated Actions API client for +// an observation cycle. Long-running watches use it before every poll so an +// OAuth token refresh is reflected by the SDK client that issues the request. +type ActionRunWatchClientFactory func() (ActionRunWatchClient, error) + // ActionRunWatchOptions configures workflow-run observation. type ActionRunWatchOptions struct { RunIDs []int64 @@ -70,12 +75,13 @@ type actionRunWatchTrackedRun struct { } type actionRunWatchSession struct { - client ActionRunWatchClient - owner string - repo string - opts ActionRunWatchOptions - emit func(ActionRunWatchEvent) error - runs map[int64]*actionRunWatchTrackedRun + clientFactory ActionRunWatchClientFactory + client ActionRunWatchClient + owner string + repo string + opts ActionRunWatchOptions + emit func(ActionRunWatchEvent) error + runs map[int64]*actionRunWatchTrackedRun } // WatchActionRuns observes runs until all are terminal, timeout, or cancellation. @@ -85,10 +91,28 @@ func WatchActionRuns( owner, repo string, opts ActionRunWatchOptions, emit func(ActionRunWatchEvent) error, +) (ActionRunWatchResult, error) { + return WatchActionRunsWithClientFactory(ctx, func() (ActionRunWatchClient, error) { + return client, nil + }, owner, repo, opts, emit) +} + +// WatchActionRunsWithClientFactory observes runs until all are terminal, +// timeout, or cancellation. It creates a client for the initial observation +// and every later polling cycle. +func WatchActionRunsWithClientFactory( + ctx stdctx.Context, + clientFactory ActionRunWatchClientFactory, + owner, repo string, + opts ActionRunWatchOptions, + emit func(ActionRunWatchEvent) error, ) (ActionRunWatchResult, error) { if err := validateActionRunWatchOptions(opts); err != nil { return ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err } + if clientFactory == nil { + return ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, fmt.Errorf("action run watch client factory is required") + } if err := ctx.Err(); err != nil { return ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err } @@ -96,7 +120,7 @@ func WatchActionRuns( watchCtx, cancel := stdctx.WithTimeout(ctx, opts.Timeout) defer cancel() - session := newActionRunWatchSession(client, owner, repo, opts, emit) + session := newActionRunWatchSessionWithClientFactory(clientFactory, owner, repo, opts, emit) done, result, err := session.initialize(watchCtx, time.Now().UTC()) if err != nil || done { return result, err @@ -151,18 +175,33 @@ func newActionRunWatchSession( owner, repo string, opts ActionRunWatchOptions, emit func(ActionRunWatchEvent) error, +) *actionRunWatchSession { + return newActionRunWatchSessionWithClientFactory(func() (ActionRunWatchClient, error) { + return client, nil + }, owner, repo, opts, emit) +} + +func newActionRunWatchSessionWithClientFactory( + clientFactory ActionRunWatchClientFactory, + owner, repo string, + opts ActionRunWatchOptions, + emit func(ActionRunWatchEvent) error, ) *actionRunWatchSession { return &actionRunWatchSession{ - client: client, - owner: owner, - repo: repo, - opts: opts, - emit: emit, - runs: make(map[int64]*actionRunWatchTrackedRun, len(opts.RunIDs)), + clientFactory: clientFactory, + owner: owner, + repo: repo, + opts: opts, + emit: emit, + runs: make(map[int64]*actionRunWatchTrackedRun, len(opts.RunIDs)), } } func (s *actionRunWatchSession) initialize(ctx stdctx.Context, now time.Time) (bool, ActionRunWatchResult, error) { + if err := s.refreshClient(); err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + headMismatch := false for _, runID := range s.opts.RunIDs { run, err := s.fetchRun(ctx, runID) @@ -215,6 +254,10 @@ func (s *actionRunWatchSession) initialize(ctx stdctx.Context, now time.Time) (b } func (s *actionRunWatchSession) poll(ctx stdctx.Context, now time.Time) (bool, ActionRunWatchResult, error) { + if err := s.refreshClient(); err != nil { + return false, ActionRunWatchResult{ExitCode: ActionRunWatchExitError}, err + } + for _, runID := range s.opts.RunIDs { previous := s.runs[runID] if isActionRunTerminal(previous.run) { @@ -288,6 +331,18 @@ func (s *actionRunWatchSession) poll(ctx stdctx.Context, now time.Time) (bool, A return false, ActionRunWatchResult{}, nil } +func (s *actionRunWatchSession) refreshClient() error { + client, err := s.clientFactory() + if err != nil { + return fmt.Errorf("failed to create action run watch client: %w", err) + } + if client == nil { + return fmt.Errorf("failed to create action run watch client: client is nil") + } + s.client = client + return nil +} + func (s *actionRunWatchSession) fetchRun(ctx stdctx.Context, runID int64) (ActionRunWatchRun, error) { run, _, err := s.client.GetRepoRun(ctx, s.owner, s.repo, runID) if err != nil { diff --git a/modules/task/actions_run_watch_test.go b/modules/task/actions_run_watch_test.go index 600bdd2e..d9bbb08e 100644 --- a/modules/task/actions_run_watch_test.go +++ b/modules/task/actions_run_watch_test.go @@ -100,6 +100,66 @@ func TestActionRunWatchNormalPollsFetchRunsOnly(t *testing.T) { assert.Equal(t, []string{"binding", "transition", "transition", "summary"}, actionRunWatchEventTypes(events)) } +func TestActionRunWatchRefreshesClientBeforeEveryPoll(t *testing.T) { + initialClient := newScriptedActionRunWatchClient() + initialClient.runs[1] = []*gitea.ActionWorkflowRun{testActionRun(1, "queued", "")} + initialClient.jobs[1] = [][]*gitea.ActionWorkflowJob{{}} + + refreshedClient := newScriptedActionRunWatchClient() + refreshedClient.runs[1] = []*gitea.ActionWorkflowRun{testActionRun(1, "completed", "success")} + refreshedClient.jobs[1] = [][]*gitea.ActionWorkflowJob{{}} + + clients := []ActionRunWatchClient{initialClient, refreshedClient} + clientCalls := 0 + session := newActionRunWatchSessionWithClientFactory(func() (ActionRunWatchClient, error) { + client := clients[clientCalls] + clientCalls++ + return client, nil + }, "gitea", "tea", testWatchOptions(), func(ActionRunWatchEvent) error { return nil }) + + t0 := time.Date(2026, time.August, 3, 10, 0, 0, 0, time.UTC) + done, _, err := session.initialize(t.Context(), t0) + require.NoError(t, err) + assert.False(t, done) + assert.Equal(t, 1, clientCalls) + assert.Equal(t, 1, initialClient.runCalls[1]) + assert.Equal(t, 1, initialClient.jobCalls[1]) + + done, result, err := session.poll(t.Context(), t0.Add(time.Minute)) + require.NoError(t, err) + assert.True(t, done) + assert.Equal(t, ActionRunWatchExitSuccess, result.ExitCode) + assert.Equal(t, 2, clientCalls) + assert.Equal(t, 1, initialClient.runCalls[1], "the initial client must not be reused after a poll starts") + assert.Equal(t, 1, refreshedClient.runCalls[1]) + assert.Equal(t, 1, refreshedClient.jobCalls[1]) +} + +func TestActionRunWatchReportsClientRefreshFailure(t *testing.T) { + client := newScriptedActionRunWatchClient() + client.runs[1] = []*gitea.ActionWorkflowRun{testActionRun(1, "queued", "")} + client.jobs[1] = [][]*gitea.ActionWorkflowJob{{}} + + clientCalls := 0 + session := newActionRunWatchSessionWithClientFactory(func() (ActionRunWatchClient, error) { + clientCalls++ + if clientCalls == 1 { + return client, nil + } + return nil, errors.New("token refresh failed") + }, "gitea", "tea", testWatchOptions(), func(ActionRunWatchEvent) error { return nil }) + + t0 := time.Date(2026, time.August, 3, 10, 0, 0, 0, time.UTC) + done, _, err := session.initialize(t.Context(), t0) + require.NoError(t, err) + assert.False(t, done) + + done, result, err := session.poll(t.Context(), t0.Add(time.Minute)) + require.ErrorContains(t, err, "token refresh failed") + assert.False(t, done) + assert.Equal(t, ActionRunWatchExitError, result.ExitCode) +} + func TestActionRunWatchStallProbeDetectsJobChangeBeforeStall(t *testing.T) { client := newScriptedActionRunWatchClient() client.runs[1] = []*gitea.ActionWorkflowRun{