From bc8161c673da8b6e72083940ffe2449513fd5d86 Mon Sep 17 00:00:00 2001 From: silverwind Date: Wed, 19 Aug 2026 11:13:21 +0000 Subject: [PATCH] fix: bound blocking calls and stop failing silently (#1174) Jobs occasionally go silent ([example](https://gitea.com/gitea/runner/actions/runs/805045/jobs/1055123)) mid-run and Gitea reaped them after `ZOMBIE_TASK_TIMEOUT`, with no error in the log. This contains a number of related fixes, all with full test coverage: 1. Bound every RPC to Gitea with a timeout, a stalled report otherwise parked logs and heartbeats for the whole job. 2. Cap `runner.fetch_timeout` at that ceiling. 3. Let only the daemon loop close its own channel, the race panicked the process. 4. Stop the job on any terminal server result, not just `RESULT_CANCELLED`. 5. Report that result instead of relabelling it as cancelled. 6. Log reporting failures once at each end of an outage instead of discarding them. 7. Clamp the acknowledged log index, a too-large ack panicked on a slice bound. 8. Stop reading server health from a `FetchTask` deadline, it marked the runner healthy and reset the error backoff on a timeout. 9. Return an error from the Docker version probe instead of a `logrus` panic. 10. Pass the context to go-git's fetch and pull. 11. Fail the clone when a refresh dies on a cancelled context. 12. Set `terminationGracePeriodSeconds` in the Kubernetes examples. Also contains a deprecation fix for goreleaser. Reviewed-on: https://gitea.com/gitea/runner/pulls/1174 Reviewed-by: bircni Co-authored-by: silverwind --- .goreleaser.yaml | 2 +- act/common/git/git.go | 17 ++- act/common/git/git_test.go | 58 ++++++++++ act/container/docker_run.go | 20 ++-- act/container/docker_run_test.go | 17 +++ examples/kubernetes/README.md | 2 + examples/kubernetes/dind-docker.yaml | 1 + examples/kubernetes/manifests_test.go | 38 +++++++ examples/kubernetes/rootless-docker.yaml | 1 + examples/kubernetes/statefulset-dind.yaml | 1 + internal/app/cmd/daemon.go | 1 + internal/app/cmd/register.go | 1 + internal/app/poll/poller.go | 24 +++- internal/app/poll/poller_test.go | 34 +++++- internal/pkg/client/http.go | 12 +- internal/pkg/client/http_test.go | 12 +- internal/pkg/config/config.example.yaml | 1 + internal/pkg/config/config.go | 7 ++ internal/pkg/config/config_test.go | 12 ++ internal/pkg/report/reporter.go | 92 +++++++++------ internal/pkg/report/reporter_test.go | 132 ++++++++++++++++++++++ 21 files changed, 422 insertions(+), 63 deletions(-) create mode 100644 examples/kubernetes/manifests_test.go diff --git a/.goreleaser.yaml b/.goreleaser.yaml index 39fa4744..5fcc4780 100644 --- a/.goreleaser.yaml +++ b/.goreleaser.yaml @@ -113,7 +113,7 @@ publishers: - R2_SECRET_ACCESS_KEY={{ index .Env "R2_SECRET_ACCESS_KEY" }} archives: - - format: binary + - formats: [binary] name_template: "{{ .Binary }}" allow_different_binary_count: true diff --git a/act/common/git/git.go b/act/common/git/git.go index fab65532..9e4ca9b6 100644 --- a/act/common/git/git.go +++ b/act/common/git/git.go @@ -345,6 +345,16 @@ func gitOptions(token string) (fetchOptions git.FetchOptions, pullOptions git.Pu return fetchOptions, pullOptions } +// staleRefreshErr reports why a failed refresh must abort: the resolve and +// checkout that follow are local and succeed on a cancelled context, which +// would hand back the cached revision as if it were fresh. +func staleRefreshErr(ctx context.Context, err error) error { + if err == nil || errors.Is(err, git.NoErrAlreadyUpToDate) { + return nil + } + return ctx.Err() +} + // NewGitCloneExecutor creates an executor to clone git repos func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor { return func(ctx context.Context) error { @@ -385,7 +395,7 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor { } if !isOfflineMode { - err = r.Fetch(&fetchOptions) + err = r.FetchContext(ctx, &fetchOptions) if err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) { return err } @@ -454,9 +464,12 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor { switch { case !isOfflineMode && !shallow: // In shallow mode the depth-limited fetch above already advanced the ref. - if err = w.Pull(&pullOptions); err != nil && err != git.NoErrAlreadyUpToDate { + if err = w.PullContext(ctx, &pullOptions); err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) { logger.Debugf("Unable to pull %s: %v", refName, err) } + if err := staleRefreshErr(ctx, err); err != nil { + return err + } case isOfflineMode && reused: reusedMsg = " (reused in offline mode)" } diff --git a/act/common/git/git_test.go b/act/common/git/git_test.go index 6984130e..40bbddf2 100644 --- a/act/common/git/git_test.go +++ b/act/common/git/git_test.go @@ -6,18 +6,24 @@ package git import ( "context" + "errors" "fmt" + "net/http" + "net/http/httptest" "os" "os/exec" "path/filepath" "strconv" "strings" + "sync" "syscall" "testing" "time" "gitea.com/gitea/runner/act/common" + gogit "github.com/go-git/go-git/v5" + gogitconfig "github.com/go-git/go-git/v5/config" log "github.com/sirupsen/logrus" logrustest "github.com/sirupsen/logrus/hooks/test" "github.com/stretchr/testify/assert" @@ -610,3 +616,55 @@ func TestAcquireCloneLock(t *testing.T) { } }) } + +// An unresponsive remote must not pin a job: the refresh has to be interruptible. +func TestNewGitCloneExecutorFetchHonoursContext(t *testing.T) { + block := make(chan struct{}) + reached := make(chan struct{}) + var once sync.Once + server := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, _ *http.Request) { + once.Do(func() { close(reached) }) + <-block + })) + t.Cleanup(func() { + close(block) + server.Close() + }) + + dir := filepath.Join(t.TempDir(), "cached-action") + repo, err := gogit.PlainInit(dir, false) + require.NoError(t, err) + _, err = repo.CreateRemote(&gogitconfig.RemoteConfig{Name: "origin", URLs: []string{server.URL}}) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + done := make(chan error, 1) + go func() { + done <- NewGitCloneExecutor(NewGitCloneExecutorInput{URL: server.URL, Ref: "main", Dir: dir})(ctx) + }() + + select { + case <-reached: + case <-time.After(10 * time.Second): + t.Fatal("the executor never reached the remote") + } + cancel() + + select { + case err := <-done: + require.Error(t, err) + case <-time.After(10 * time.Second): + t.Fatal("fetch ignored context cancellation") + } +} + +func TestStaleRefreshErr(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + require.NoError(t, staleRefreshErr(ctx, errors.New("remote hung up"))) + + cancel() + require.ErrorIs(t, staleRefreshErr(ctx, errors.New("remote hung up")), context.Canceled) + require.NoError(t, staleRefreshErr(ctx, gogit.NoErrAlreadyUpToDate)) +} diff --git a/act/container/docker_run.go b/act/container/docker_run.go index 20125229..00e5ad40 100644 --- a/act/container/docker_run.go +++ b/act/container/docker_run.go @@ -88,14 +88,14 @@ func (cr *containerReference) connectToNetwork(name string, aliases []string) co } } -// supportsContainerImagePlatform returns true if the underlying Docker server -// API version is 1.41 and beyond -func supportsContainerImagePlatform(ctx context.Context, cli client.APIClient) bool { +// supportsContainerImagePlatform reports whether the Docker server API version +// is 1.41 and beyond +func supportsContainerImagePlatform(ctx context.Context, cli client.APIClient) (bool, error) { ver, err := cli.ServerVersion(ctx, client.ServerVersionOptions{}) if err != nil { - common.Logger(ctx).Panicf("Failed to get Docker API Version: %s", err) + return false, fmt.Errorf("get docker API version: %w", err) } - return versions.GreaterThanOrEqualTo(ver.APIVersion, "1.41") + return versions.GreaterThanOrEqualTo(ver.APIVersion, "1.41"), nil } func (cr *containerReference) Create(capAdd, capDrop []string) common.Executor { @@ -682,11 +682,17 @@ func (cr *containerReference) create(capAdd, capDrop []string) common.Executor { } var platSpecs *specs.Platform - if cr.input.Platform != "" && supportsContainerImagePlatform(ctx, cr.cli) { - platSpecs, err = parsePlatform(cr.input.Platform) + if cr.input.Platform != "" { + // Dropping the platform silently would build for the host arch. + supported, err := supportsContainerImagePlatform(ctx, cr.cli) if err != nil { return err } + if supported { + if platSpecs, err = parsePlatform(cr.input.Platform); err != nil { + return err + } + } } hostConfig := &container.HostConfig{ diff --git a/act/container/docker_run_test.go b/act/container/docker_run_test.go index 9ff339fa..8a1eaed8 100644 --- a/act/container/docker_run_test.go +++ b/act/container/docker_run_test.go @@ -79,6 +79,11 @@ type mockDockerClient struct { mock.Mock } +func (m *mockDockerClient) ServerVersion(ctx context.Context, opts mobyclient.ServerVersionOptions) (mobyclient.ServerVersionResult, error) { + args := m.Called(ctx, opts) + return args.Get(0).(mobyclient.ServerVersionResult), args.Error(1) +} + func (m *mockDockerClient) ExecCreate(ctx context.Context, id string, opts mobyclient.ExecCreateOptions) (mobyclient.ExecCreateResult, error) { args := m.Called(ctx, id, opts) return args.Get(0).(mobyclient.ExecCreateResult), args.Error(1) @@ -930,3 +935,15 @@ func TestMergeContainerConfigsVolumesReplaceRunnerMounts(t *testing.T) { assert.Equal(t, []string{"/var/run/docker.sock:/var/run/docker.sock", "/host/tools:/opt/hostedtoolcache"}, hostConf.Binds) assert.Empty(t, hostConf.Mounts) } + +// A dead daemon must fail the job, not panic through logrus and not silently +// drop the requested platform. +func TestSupportsContainerImagePlatformDaemonError(t *testing.T) { + cli := &mockDockerClient{} + cli.On("ServerVersion", mock.Anything, mock.Anything). + Return(mobyclient.ServerVersionResult{}, errors.New("cannot connect to the Docker daemon")) + + supported, err := supportsContainerImagePlatform(t.Context(), cli) + require.ErrorContains(t, err, "cannot connect to the Docker daemon") + assert.False(t, supported) +} diff --git a/examples/kubernetes/README.md b/examples/kubernetes/README.md index b7ddada5..d79055a6 100644 --- a/examples/kubernetes/README.md +++ b/examples/kubernetes/README.md @@ -11,6 +11,8 @@ Each example persists **two** things, and it is worth knowing which is which: - `/data` is the runner's working directory. It holds the `.runner` registration file and, optionally, the config file — so the runner re-attaches to the server instead of registering again. - The Docker daemon's data root holds the images pulled for jobs (`/var/lib/docker` for the dind sidecar, `/home/rootless/.local/share/docker` for `dind-rootless`). It is *not* under `/data`. If you drop this volume, the examples still work, but the image cache is discarded whenever the pod is recreated and every job re-pulls its images. +- Kubernetes SIGKILLs a pod 30s after SIGTERM by default, long before a job finishes and reports its result, which leaves tasks the server can only reap as zombies. The manifests raise `terminationGracePeriodSeconds` to three hours, matching the systemd example and the `runner.timeout` job ceiling; set `runner.shutdown_timeout` below that so the runner drains jobs within the window rather than being killed mid-cleanup. + Files in this directory: - [`dind-docker.yaml`](dind-docker.yaml) diff --git a/examples/kubernetes/dind-docker.yaml b/examples/kubernetes/dind-docker.yaml index cfe6eb5d..75e02a43 100644 --- a/examples/kubernetes/dind-docker.yaml +++ b/examples/kubernetes/dind-docker.yaml @@ -56,6 +56,7 @@ spec: app: runner spec: restartPolicy: Always + terminationGracePeriodSeconds: 10800 # keep above runner.shutdown_timeout, see README volumes: - name: docker-socket emptyDir: {} diff --git a/examples/kubernetes/manifests_test.go b/examples/kubernetes/manifests_test.go new file mode 100644 index 00000000..a37b103d --- /dev/null +++ b/examples/kubernetes/manifests_test.go @@ -0,0 +1,38 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package kubernetes_test + +import ( + "os" + "path/filepath" + "regexp" + "strconv" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +var gracePeriod = regexp.MustCompile(`terminationGracePeriodSeconds: (\d+)`) + +// Without it Kubernetes SIGKILLs the pod 30s after SIGTERM, mid-job. +func TestManifestsSetTerminationGracePeriod(t *testing.T) { + files, err := filepath.Glob("*.yaml") + require.NoError(t, err) + require.NotEmpty(t, files) + + for _, file := range files { + content, err := os.ReadFile(file) + require.NoError(t, err) + if !strings.Contains(string(content), "containers:") { + continue + } + match := gracePeriod.FindStringSubmatch(string(content)) + require.NotNil(t, match, file) + seconds, err := strconv.Atoi(match[1]) + require.NoError(t, err) + assert.GreaterOrEqual(t, seconds, 3600, file) + } +} diff --git a/examples/kubernetes/rootless-docker.yaml b/examples/kubernetes/rootless-docker.yaml index 62f5539c..43ef81fc 100644 --- a/examples/kubernetes/rootless-docker.yaml +++ b/examples/kubernetes/rootless-docker.yaml @@ -56,6 +56,7 @@ spec: app: runner spec: restartPolicy: Always + terminationGracePeriodSeconds: 10800 # keep above runner.shutdown_timeout, see README volumes: - name: runner-data persistentVolumeClaim: diff --git a/examples/kubernetes/statefulset-dind.yaml b/examples/kubernetes/statefulset-dind.yaml index ad8116a7..2d3c6cfa 100644 --- a/examples/kubernetes/statefulset-dind.yaml +++ b/examples/kubernetes/statefulset-dind.yaml @@ -33,6 +33,7 @@ spec: app: runner spec: restartPolicy: Always + terminationGracePeriodSeconds: 10800 # keep above runner.shutdown_timeout, see README volumes: - name: docker-socket emptyDir: {} diff --git a/internal/app/cmd/daemon.go b/internal/app/cmd/daemon.go index d1b3a69a..b70463c7 100644 --- a/internal/app/cmd/daemon.go +++ b/internal/app/cmd/daemon.go @@ -158,6 +158,7 @@ func runDaemon(ctx context.Context, daemArgs *daemonArgs, configFile *string) fu cfg.Runner.Insecure, reg.UUID, reg.Token, + config.RequestTimeout, ) runner := run.NewRunner(cfg, reg, cli) diff --git a/internal/app/cmd/register.go b/internal/app/cmd/register.go index 2de43b8e..59a36254 100644 --- a/internal/app/cmd/register.go +++ b/internal/app/cmd/register.go @@ -366,6 +366,7 @@ func doRegister(ctx context.Context, cfg *config.Config, inputs *registerInputs) cfg.Runner.Insecure, "", "", + config.RequestTimeout, ) for { diff --git a/internal/app/poll/poller.go b/internal/app/poll/poller.go index 13e2ffde..95423e96 100644 --- a/internal/app/poll/poller.go +++ b/internal/app/poll/poller.go @@ -68,6 +68,8 @@ type Poller struct { type workerState struct { consecutiveEmpty int64 consecutiveErrors int64 + // fetchTimedOut suppresses repeats of the fetch timeout warning. + fetchTimedOut bool // lastBackoff is the last interval reported to the PollBackoffSeconds gauge; // used to suppress redundant no-op Set calls when the backoff plateaus // (e.g. at FetchIntervalMax). @@ -346,16 +348,20 @@ func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task, TasksVersion: v, })) - // DeadlineExceeded is the designed idle path for a long-poll: the server - // found no work within FetchTimeout. Treat it as an empty response and do - // not record the duration — the timeout value would swamp the histogram. + // Our own deadline proves nothing either way: today Gitea answers immediately, + // so it means a slow server, and once it holds the request open it is an idle + // poll. Back off without claiming the server is healthy, warn once a streak so + // neither case floods, and keep it out of the latency histogram. if errors.Is(err, context.DeadlineExceeded) { - p.markHealthyPoll() + if !s.fetchTimedOut { + s.fetchTimedOut = true + log.Warnf("fetching a task timed out after %s, raise runner.fetch_timeout if this persists", p.cfg.Runner.FetchTimeout) + } s.consecutiveEmpty++ - s.consecutiveErrors = 0 // timeout is a healthy idle response metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultEmpty).Inc() return nil, false } + s.fetchTimedOut = false metrics.PollFetchDuration.Observe(time.Since(start).Seconds()) if err != nil { @@ -368,7 +374,13 @@ func (p *Poller) fetchTask(ctx context.Context, s *workerState) (*runnerv1.Task, p.shutdownPolling() return nil, false } - log.WithError(err).Error("failed to fetch task") + // Not a long poll, so a deadline can mean the server assigned a task + // this runner never received. + if errors.Is(err, context.DeadlineExceeded) { + log.WithError(err).Errorf("fetching a task timed out after %s", p.cfg.Runner.FetchTimeout) + } else { + log.WithError(err).Error("failed to fetch task") + } p.lastPollFailed.Store(true) s.consecutiveErrors++ metrics.PollFetchTotal.WithLabelValues(metrics.LabelResultError).Inc() diff --git a/internal/app/poll/poller_test.go b/internal/app/poll/poller_test.go index 2129e143..33dcb482 100644 --- a/internal/app/poll/poller_test.go +++ b/internal/app/poll/poller_test.go @@ -16,6 +16,7 @@ import ( connect_go "connectrpc.com/connect" runnerv1 "gitea.dev/actionslib/runner/v1" + "github.com/sirupsen/logrus/hooks/test" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -61,11 +62,7 @@ func TestPoller_WorkerStateCounters(t *testing.T) { // increments only the per-worker error counter, not the empty counter. func TestPoller_FetchErrorIncrementsErrorsOnly(t *testing.T) { client := mocks.NewClient(t) - client.On("FetchTask", mock.Anything, mock.Anything).Return( - func(_ context.Context, _ *connect_go.Request[runnerv1.FetchTaskRequest]) (*connect_go.Response[runnerv1.FetchTaskResponse], error) { - return nil, errors.New("network unreachable") - }, - ) + client.On("FetchTask", mock.Anything, mock.Anything).Return(nil, errors.New("network unreachable")) cfg, err := config.LoadDefault("") require.NoError(t, err) @@ -78,6 +75,33 @@ func TestPoller_FetchErrorIncrementsErrorsOnly(t *testing.T) { assert.Equal(t, int64(0), s.consecutiveEmpty) } +// A deadline is the idle path once the server holds the request open, and a +// symptom before then, so it must neither reset the error count nor assert that +// the server is reachable. +func TestPoller_FetchTimeoutIsNoSignal(t *testing.T) { + client := mocks.NewClient(t) + client.On("FetchTask", mock.Anything, mock.Anything).Return(nil, context.DeadlineExceeded) + + cfg, err := config.LoadDefault("") + require.NoError(t, err) + p := &Poller{client: client, cfg: cfg} + p.lastPollFailed.Store(true) + + hook := test.NewGlobal() + defer hook.Reset() + + s := &workerState{consecutiveErrors: 2} + _, ok := p.fetchTask(context.Background(), s) + require.False(t, ok) + assert.Equal(t, int64(2), s.consecutiveErrors) + assert.Equal(t, int64(1), s.consecutiveEmpty) + assert.True(t, p.lastPollFailed.Load(), "a timeout must not clear a known failure") + + // An idle runner against a server holding the request open must not flood. + _, _ = p.fetchTask(context.Background(), s) + assert.Len(t, hook.AllEntries(), 1) +} + // TestPoller_FetchUnauthenticatedStopsPolling verifies that an Unauthenticated // response marks the runner as unregistered and cancels the polling context so // the daemon can exit instead of retrying forever. diff --git a/internal/pkg/client/http.go b/internal/pkg/client/http.go index 57fd0cd0..ada7a48e 100644 --- a/internal/pkg/client/http.go +++ b/internal/pkg/client/http.go @@ -17,7 +17,7 @@ import ( "gitea.dev/actionslib/runner/v1/runnerv1connect" ) -func getHTTPClient(endpoint string, insecure bool) *http.Client { +func getHTTPClient(endpoint string, insecure bool, timeout time.Duration) *http.Client { transport := &http.Transport{ Proxy: http.ProxyFromEnvironment, MaxIdleConns: 10, @@ -29,11 +29,13 @@ func getHTTPClient(endpoint string, insecure bool) *http.Client { InsecureSkipVerify: true, } } - return &http.Client{Transport: transport} + return &http.Client{Transport: transport, Timeout: timeout} } -// New returns a new runner client. -func New(endpoint string, insecure bool, uuid, token string, opts ...connect.ClientOption) *HTTPClient { +// New returns a new runner client. timeout bounds every RPC: without it a +// stalled connection parks the reporter for the whole job context, so logs and +// heartbeats stop together and the task is reaped as a zombie. +func New(endpoint string, insecure bool, uuid, token string, timeout time.Duration, opts ...connect.ClientOption) *HTTPClient { baseURL := strings.TrimRight(endpoint, "/") + "/api/actions" opts = append(opts, connect.WithInterceptors(connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc { @@ -49,7 +51,7 @@ func New(endpoint string, insecure bool, uuid, token string, opts ...connect.Cli } }))) - httpClient := getHTTPClient(endpoint, insecure) + httpClient := getHTTPClient(endpoint, insecure, timeout) return &HTTPClient{ PingServiceClient: pingv1connect.NewPingServiceClient( httpClient, diff --git a/internal/pkg/client/http_test.go b/internal/pkg/client/http_test.go index 866aaa93..ef555dc0 100644 --- a/internal/pkg/client/http_test.go +++ b/internal/pkg/client/http_test.go @@ -8,6 +8,7 @@ import ( "net/http/httptest" "strings" "testing" + "time" "connectrpc.com/connect" pingv1 "gitea.dev/actionslib/ping/v1" @@ -17,7 +18,8 @@ import ( func TestGetHTTPClientUsesProxyFromEnvironment(t *testing.T) { t.Setenv("HTTP_PROXY", "http://proxy.example.com:8080") - client := getHTTPClient("http://gitea.example.com", false) + client := getHTTPClient("http://gitea.example.com", false, time.Minute) + require.Equal(t, time.Minute, client.Timeout) transport, ok := client.Transport.(*http.Transport) require.True(t, ok) @@ -32,7 +34,7 @@ func TestGetHTTPClientUsesProxyFromEnvironment(t *testing.T) { func TestGetHTTPClientInsecureTLS(t *testing.T) { // insecure only takes effect for https endpoints - httpsInsecure := getHTTPClient("https://gitea.example.com", true) + httpsInsecure := getHTTPClient("https://gitea.example.com", true, time.Minute) transport, ok := httpsInsecure.Transport.(*http.Transport) require.True(t, ok) require.NotNil(t, transport.TLSClientConfig) @@ -47,7 +49,7 @@ func TestGetHTTPClientInsecureTLS(t *testing.T) { {"http insecure ignored", "http://gitea.example.com", true}, } { t.Run(tc.name, func(t *testing.T) { - c := getHTTPClient(tc.endpoint, tc.insecure) + c := getHTTPClient(tc.endpoint, tc.insecure, time.Minute) tr, ok := c.Transport.(*http.Transport) require.True(t, ok) require.Nil(t, tr.TLSClientConfig) @@ -66,7 +68,7 @@ func TestNewSetsBaseURLAndHeaders(t *testing.T) { defer server.Close() // trailing slash must be trimmed before "/api/actions" is appended - c := New(server.URL+"/", false, "the-uuid", "the-token") + c := New(server.URL+"/", false, "the-uuid", "the-token", time.Minute) // Address returns the endpoint as supplied (untrimmed) require.Equal(t, server.URL+"/", c.Address()) require.False(t, c.Insecure()) @@ -87,7 +89,7 @@ func TestNewOmitsEmptyHeaders(t *testing.T) { })) defer server.Close() - c := New(server.URL, false, "", "") + c := New(server.URL, false, "", "", time.Minute) _, _ = c.Ping(t.Context(), connect.NewRequest(&pingv1.PingRequest{Data: "hi"})) require.Empty(t, gotHeaders.Get(UUIDHeader)) diff --git a/internal/pkg/config/config.example.yaml b/internal/pkg/config/config.example.yaml index eab75350..e5aea4e3 100644 --- a/internal/pkg/config/config.example.yaml +++ b/internal/pkg/config/config.example.yaml @@ -34,6 +34,7 @@ runner: # Whether skip verifying the TLS certificate of the Gitea instance. #insecure: false # The timeout for fetching the job from the Gitea instance. + # Values above the 60s RPC timeout are capped to it. #fetch_timeout: 5s # The interval for fetching the job from the Gitea instance. #fetch_interval: 2s diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index 52e509d6..6a71c898 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -20,6 +20,9 @@ import ( "go.yaml.in/yaml/v4" ) +// RequestTimeout bounds every RPC to Gitea, and with it runner.fetch_timeout. +const RequestTimeout = 60 * time.Second + // DefaultPostTaskScriptTimeout is the fallback cap on how long the post-task // script may run when post_task_script is set without an explicit timeout. It is // applied both at config load (for a configured script) and at the point of use @@ -334,6 +337,10 @@ func LoadDefault(file string) (*Config, error) { log.Warnf("runner.tool_cache_mode %q with capacity %d: two jobs writing the same tool version at once corrupt it", ToolCacheModeShared, cfg.Runner.Capacity) } + if cfg.Runner.FetchTimeout > RequestTimeout { + log.Warnf("fetch_timeout (%v) exceeds the RPC timeout (%v), capping it", cfg.Runner.FetchTimeout, RequestTimeout) + cfg.Runner.FetchTimeout = RequestTimeout + } if cfg.Runner.FetchIntervalMax < cfg.Runner.FetchInterval { log.Warnf("fetch_interval_max (%v) is less than fetch_interval (%v), setting fetch_interval_max to fetch_interval", cfg.Runner.FetchIntervalMax, cfg.Runner.FetchInterval) diff --git a/internal/pkg/config/config_test.go b/internal/pkg/config/config_test.go index 56b66ef5..bd860357 100644 --- a/internal/pkg/config/config_test.go +++ b/internal/pkg/config/config_test.go @@ -416,3 +416,15 @@ func TestLoadDefault_ShippedConfigsChangeNothing(t *testing.T) { } assert.Empty(t, hook.AllEntries()) } + +func TestLoadDefault_ClampsFetchTimeout(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.yaml") + require.NoError(t, os.WriteFile(path, []byte(` +runner: + fetch_timeout: 120s +`), 0o600)) + + cfg, err := LoadDefault(path) + require.NoError(t, err) + assert.Equal(t, RequestTimeout, cfg.Runner.FetchTimeout) +} diff --git a/internal/pkg/report/reporter.go b/internal/pkg/report/reporter.go index ecfe3c71..133b5f07 100644 --- a/internal/pkg/report/reporter.go +++ b/internal/pkg/report/reporter.go @@ -26,6 +26,9 @@ import ( "google.golang.org/protobuf/types/known/timestamppb" ) +// errOutputsNotSent travels the same return path as transport failures but is not one. +var errOutputsNotSent = errors.New("there are still outputs that have not been sent") + // Size limits for the outputs reported to the server. const ( maxOutputKeyLen = 255 @@ -56,8 +59,13 @@ type Reporter struct { // so the gauge skips no-op Set calls when the buffer size is unchanged. lastLogBufferRows int - state *runnerv1.TaskState - stateChanged bool + state *runnerv1.TaskState + stateChanged bool + // reportFailing keeps an outage to one log line at each end. + reportFailing map[string]bool + // serverResult is what the server decided, e.g. the zombie reaper failing + // the task. Guarded by stateMu. + serverResult runnerv1.Result stateMu sync.RWMutex outputsMu sync.Mutex outputs map[string]jobOutput @@ -77,6 +85,8 @@ type Reporter struct { // closeTimeout bounds each RPC attempt in the final flush, on a context // detached from r.ctx so a server cancel can't abort the acknowledgement. closeTimeout time.Duration + // daemonWait bounds how long Close waits for the daemon loop to acknowledge. + daemonWait time.Duration // Event notification channels (non-blocking, buffered 1) logNotify chan struct{} // signal: new log rows arrived @@ -119,10 +129,13 @@ func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.C state: &runnerv1.TaskState{ Id: task.Id, }, + reportFailing: map[string]bool{}, daemon: make(chan struct{}), heartbeatStop: make(chan struct{}), } + rv.daemonWait = 6 * rv.effectiveCloseTimeout() + if task.Secrets["ACTIONS_STEP_DEBUG"] == "true" { rv.debugOutputEnabled = true } @@ -293,6 +306,21 @@ func (r *Reporter) Fire(entry *log.Entry) error { return nil } +// Only the daemon loop calls this, so reportFailing needs no lock. +func (r *Reporter) noteReport(method string, err error) { + if errors.Is(err, errOutputsNotSent) { + err = nil // the RPC itself succeeded + } + switch { + case err != nil && !r.reportFailing[method]: + r.reportFailing[method] = true + log.Warnf("%s error: %v, retrying until reconnected", method, err) + case err == nil && r.reportFailing[method]: + delete(r.reportFailing, method) + log.Infof("%s reconnected", method) + } +} + func (r *Reporter) RunDaemon() { go r.runDaemonLoop() } @@ -319,6 +347,8 @@ func (r *Reporter) stopLatencyTimer(active *bool, timer *time.Timer) { } func (r *Reporter) runDaemonLoop() { + defer close(r.daemon) + logTicker := time.NewTicker(r.logReportInterval) stateTicker := time.NewTicker(r.stateReportInterval) @@ -337,11 +367,11 @@ func (r *Reporter) runDaemonLoop() { for { select { case <-logTicker.C: - _ = r.ReportLog(false) + r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false)) r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer) case <-stateTicker.C: - _ = r.ReportState(false) + r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false)) case <-r.logNotify: r.stateMu.RLock() @@ -349,7 +379,7 @@ func (r *Reporter) runDaemonLoop() { r.stateMu.RUnlock() if n >= r.logBatchSize { - _ = r.ReportLog(false) + r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false)) r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer) } else if !maxLatencyActive && n > 0 { maxLatencyTimer.Reset(r.logReportMaxLatency) @@ -358,25 +388,23 @@ func (r *Reporter) runDaemonLoop() { case <-r.stateNotify: // Step transition or job result — flush both immediately for frontend UX. - _ = r.ReportLog(false) - _ = r.ReportState(false) + r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false)) + r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false)) r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer) case <-maxLatencyTimer.C: maxLatencyActive = false - _ = r.ReportLog(false) + r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false)) case <-r.ctx.Done(): // Stop heartbeating on cancel so Gitea sees the runner as offline // during cleanup and won't assign an overlapping task. Close() still // delivers the final flush on a detached context (flushFinal). - close(r.daemon) return case <-r.heartbeatStop: // Stop heartbeating during post-task script execution. Close() still // delivers the final flush on a detached context (flushFinal). - close(r.daemon) return } @@ -384,7 +412,6 @@ func (r *Reporter) runDaemonLoop() { closed := r.closed r.stateMu.RUnlock() if closed { - close(r.daemon) return } } @@ -435,28 +462,23 @@ func (r *Reporter) Close(lastWords string) error { r.stateMu.Lock() r.closed = true if r.state.Result == runnerv1.Result_RESULT_UNSPECIFIED { - // When r.ctx has been cancelled (server returned RESULT_CANCELLED via - // rpcCtx/ReportState, see line 590) the job is being torn down on the - // cancellation path: surface that explicitly instead of attributing it - // to a generic failure. - cancelled := errors.Is(r.ctx.Err(), context.Canceled) + // No result of its own, so say why it stopped. + result, words := runnerv1.Result_RESULT_FAILURE, "Early termination" + switch { + case r.serverResult != runnerv1.Result_RESULT_UNSPECIFIED: + result, words = r.serverResult, "Ended by the server" + case errors.Is(r.ctx.Err(), context.Canceled): + result, words = runnerv1.Result_RESULT_CANCELLED, "Cancelled" + } if lastWords == "" { - if cancelled { - lastWords = "Cancelled" - } else { - lastWords = "Early termination" - } + lastWords = words } for _, v := range r.state.Steps { if v.Result == runnerv1.Result_RESULT_UNSPECIFIED { v.Result = runnerv1.Result_RESULT_CANCELLED } } - if cancelled { - r.state.Result = runnerv1.Result_RESULT_CANCELLED - } else { - r.state.Result = runnerv1.Result_RESULT_FAILURE - } + r.state.Result = result r.logRows = append(r.logRows, &runnerv1.LogRow{ Time: timestamppb.Now(), Content: lastWords, @@ -476,9 +498,8 @@ func (r *Reporter) Close(lastWords string) error { // Wait for Acknowledge select { case <-r.daemon: - case <-time.After(60 * time.Second): - close(r.daemon) - log.Error("No Response from RunDaemon for 60s, continue best effort") + case <-time.After(r.daemonWait): + log.Errorf("No Response from RunDaemon for %s, continue best effort", r.daemonWait) } // Gitea's UpdateLog short-circuits on len(Rows)==0 before honoring NoMore, @@ -569,8 +590,10 @@ func (r *Reporter) ReportLog(noMore bool) error { } r.stateMu.Lock() - r.logRows = r.logRows[ack-r.logOffset:] submitted := r.logOffset + len(rows) + // A server can ack beyond what it was sent; clamp to stay within the buffer. + ack = min(ack, submitted) + r.logRows = r.logRows[ack-r.logOffset:] r.logOffset = ack remaining := len(r.logRows) r.stateMu.Unlock() @@ -655,11 +678,16 @@ func (r *Reporter) ReportState(reportResult bool) error { } r.outputsMu.Unlock() - if resp.Msg.State != nil && resp.Msg.State.Result == runnerv1.Result_RESULT_CANCELLED { + // A terminal result means the server is done with this task; keep running and + // the job holds a capacity slot until runner.timeout. + if state := resp.Msg.State; state != nil && state.Result != runnerv1.Result_RESULT_UNSPECIFIED { + r.stateMu.Lock() + r.serverResult = state.Result + r.stateMu.Unlock() r.cancel() } if len(noSent) > 0 { - return fmt.Errorf("there are still outputs that have not been sent: %v", noSent) + return fmt.Errorf("%w: %v", errOutputsNotSent, noSent) } return nil diff --git a/internal/pkg/report/reporter_test.go b/internal/pkg/report/reporter_test.go index 8f91fddc..cec0dd23 100644 --- a/internal/pkg/report/reporter_test.go +++ b/internal/pkg/report/reporter_test.go @@ -12,6 +12,7 @@ import ( "net/url" "slices" "strings" + "sync" "sync/atomic" "testing" "time" @@ -19,10 +20,12 @@ import ( "gitea.com/gitea/runner/act/runner" "gitea.com/gitea/runner/internal/pkg/client/mocks" "gitea.com/gitea/runner/internal/pkg/config" + "gitea.com/gitea/runner/internal/pkg/metrics" connect_go "connectrpc.com/connect" runnerv1 "gitea.dev/actionslib/runner/v1" log "github.com/sirupsen/logrus" + logrustest "github.com/sirupsen/logrus/hooks/test" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -1169,3 +1172,132 @@ func TestReporter_masksEncodedSecrets(t *testing.T) { assert.NotContains(t, row.Content, base64.StdEncoding.EncodeToString([]byte(secret))) } } + +// A server that acknowledges more rows than were sent must not take the runner +// process down with it. +func TestReporter_AckIndexBeyondBuffer(t *testing.T) { + client := mocks.NewClient(t) + client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { + return connect_go.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: 1000}), nil + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + taskCtx, err := structpb.NewStruct(map[string]any{}) + require.NoError(t, err) + cfg, _ := config.LoadDefault("") + reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) + reporter.ResetSteps(1) + + require.NoError(t, reporter.Fire(&log.Entry{ + Message: "hello", + Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true}, + Level: log.InfoLevel, + })) + + require.NoError(t, reporter.ReportLog(false)) +} + +// Close has to survive giving up on a daemon still parked in an RPC. +func TestReporter_CloseWithStuckDaemon(t *testing.T) { + release := make(chan struct{}) + entered := make(chan struct{}) + var once sync.Once + + client := mocks.NewClient(t) + client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { + once.Do(func() { + close(entered) + <-release + }) + return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ + AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), + }), nil + }) + client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, _ *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { + return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{}), nil + }).Maybe() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + taskCtx, err := structpb.NewStruct(map[string]any{}) + require.NoError(t, err) + cfg, _ := config.LoadDefault("") + cfg.Runner.LogReportInterval = 10 * time.Millisecond + reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) + reporter.daemonWait = time.Millisecond + reporter.RunDaemon() + reporter.ResetSteps(1) + + require.NoError(t, reporter.Fire(&log.Entry{ + Message: "hello", + Data: log.Fields{"stage": "Main", "stepNumber": 0, "raw_output": true}, + Level: log.InfoLevel, + })) + + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("daemon never reached UpdateLog") + } + go func() { + time.Sleep(10 * reporter.daemonWait) + close(release) + }() + + require.NoError(t, reporter.Close("")) +} + +// The zombie reaper marks a task failed, and cancelling the context is how the +// runner stops, so the two must not be conflated into "cancelled". +func TestReporter_ServerFailureResultIsReportedAsFailure(t *testing.T) { + var lastState *runnerv1.TaskState + client := mocks.NewClient(t) + client.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { + lastState = req.Msg.State + return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{ + State: &runnerv1.TaskState{Result: runnerv1.Result_RESULT_FAILURE}, + }), nil + }) + client.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect_go.Request[runnerv1.UpdateLogRequest]) (*connect_go.Response[runnerv1.UpdateLogResponse], error) { + return connect_go.NewResponse(&runnerv1.UpdateLogResponse{ + AckIndex: req.Msg.Index + int64(len(req.Msg.Rows)), + }), nil + }).Maybe() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + taskCtx, err := structpb.NewStruct(map[string]any{}) + require.NoError(t, err) + cfg, _ := config.LoadDefault("") + reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) + close(reporter.daemon) // no daemon loop to acknowledge + + require.NoError(t, reporter.ReportState(false)) + require.ErrorIs(t, ctx.Err(), context.Canceled) + require.NoError(t, reporter.Close("")) + + require.NotNil(t, lastState) + assert.Equal(t, runnerv1.Result_RESULT_FAILURE, lastState.Result) +} + +func TestReporter_NoteReport(t *testing.T) { + hook := logrustest.NewGlobal() + defer hook.Reset() + + reporter := &Reporter{reportFailing: map[string]bool{}} + + // A pending output is not a transport failure and must not read as one. + reporter.noteReport(metrics.LabelMethodUpdateTask, fmt.Errorf("wrapped: %w", errOutputsNotSent)) + assert.Empty(t, hook.AllEntries()) + + // An outage logs once at each end, not every report interval for hours. + reporter.noteReport(metrics.LabelMethodUpdateTask, errors.New("connection refused")) + reporter.noteReport(metrics.LabelMethodUpdateTask, errors.New("connection refused")) + require.Len(t, hook.AllEntries(), 1) + assert.Contains(t, hook.LastEntry().Message, "connection refused") + + reporter.noteReport(metrics.LabelMethodUpdateTask, nil) + require.Len(t, hook.AllEntries(), 2) + assert.Contains(t, hook.LastEntry().Message, "reconnected") +}