utils-code/gitea.tea
1
0
Fork 0
mirror of https://gitea.com/gitea/tea.git synced 2026-09-10 07:26:33 -04:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

fix(actions): refresh watch client before polls

Browse source
This commit is contained in:
Noel 2026-08-07 20:05:12 +02:00
parent 28cf63b22b
commit 59d8bbcde1
No known key found for this signature in database
GPG key ID: ACB59B76E29853EA
3 changed files with 131 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
cmd/actions/runs/watch.go
View file

@ -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"),

81
modules/task/actions_run_watch.go
View file

@ -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 {

60
modules/task/actions_run_watch_test.go
View file

@ -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{