From 3f7fd16ea1dfdd98b02dcdf3101daff8218b26a2 Mon Sep 17 00:00:00 2001 From: silverwind Date: Thu, 30 Jul 2026 07:09:53 +0000 Subject: [PATCH] fix: fix panics, enable the `forcetypeassert` lint (#1123) An unchecked type assertion panics on input it did not expect, as a missing `tool_cache` key did in https://gitea.com/gitea/runner/pulls/1122. Every flagged site is now handled where it can fail, or typed so it cannot: a `lock.Keyed` replaces the two `sync.Map` mutex registries, and the reporter's outputs carry an explicit sent flag. Mocks keep their assertions, a mismatch there is a setup error the panic names. Bugs it turned up (only the first is reachable from workflows): 1. A scalar `matrix.include` or `matrix.exclude`, e.g. `include: foo` or `include: [1, 2]`, panicked the runner with `interface conversion: interface {} is string, not map[string]interface {}`. Verified against `main`, it is now a workflow error. `OnSchedule` panicked the same way on a malformed `on.schedule` entry. 1. `ExternalURL()` panicked on the nil listener after `Close()`, the port is now resolved once at startup. 1. `errors.Is(err, git.ErrShortRef)` followed by `err.(*git.Error)` panics as soon as anything wraps that error, so it is `errors.As` now. 1. An output name the server acknowledged without ever being sent one was recorded as sent forever, which silently dropped a later value for that name. https://gitea.com/gitea/runner/commit/48576ab3e5c485af5a21ee6d21888a049ecef42b fixes one discovered issue: a matrix key holding a nested object was logged and then run as if the job had no matrix, so it now fails like an unknown `exclude` key. Reviewed-on: https://gitea.com/gitea/runner/pulls/1123 Reviewed-by: Lunny Xiao Co-authored-by: silverwind --- .golangci.yml | 4 ++ act/artifactcache/handler.go | 11 +++- act/common/git/git.go | 8 +-- act/common/git/git_test.go | 9 --- act/common/outbound_ip.go | 4 +- act/container/docker_create_flags_test.go | 3 +- act/exprparser/interpreter_test.go | 5 +- act/model/workflow.go | 79 ++++++++++++----------- act/model/workflow_test.go | 30 +++++---- act/runner/action.go | 16 +++-- act/runner/run_context.go | 9 +-- act/runner/step_action_remote.go | 5 +- internal/pkg/lock/keyed.go | 38 +++++++++++ internal/pkg/report/reporter.go | 54 ++++++++++------ internal/pkg/report/reporter_test.go | 44 ++++++++++--- 15 files changed, 210 insertions(+), 109 deletions(-) create mode 100644 internal/pkg/lock/keyed.go diff --git a/.golangci.yml b/.golangci.yml index 42e2d5e0..8114921f 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -11,6 +11,7 @@ linters: - dupl - errcheck - forbidigo + - forcetypeassert - gocheckcompilerdirectives - gocritic - goheader @@ -102,6 +103,9 @@ linters: - linters: - forbidigo path: cmd + - linters: + - forcetypeassert + path: _test\.go issues: max-issues-per-linter: 0 max-same-issues: 0 diff --git a/act/artifactcache/handler.go b/act/artifactcache/handler.go index 6efff636..c9fb71b9 100644 --- a/act/artifactcache/handler.go +++ b/act/artifactcache/handler.go @@ -70,6 +70,7 @@ type Handler struct { storage *Storage router *httprouter.Router listener net.Listener + port int server *http.Server logger logrus.FieldLogger @@ -177,6 +178,12 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo if err != nil { return nil, err } + addr, ok := listener.Addr().(*net.TCPAddr) + if !ok { + listener.Close() + return nil, fmt.Errorf("cache server listens on %T, want a TCP address", listener.Addr()) + } + h.port = addr.Port server := &http.Server{ ReadHeaderTimeout: 2 * time.Second, Handler: router, @@ -194,9 +201,7 @@ func StartHandler(dir, outboundIP string, port uint16, internalSecret string, lo func (h *Handler) ExternalURL() string { // TODO: make the external url configurable if necessary - return fmt.Sprintf("http://%s:%d", - h.outboundIP, - h.listener.Addr().(*net.TCPAddr).Port) + return fmt.Sprintf("http://%s:%d", h.outboundIP, h.port) } // RegisterJob makes token a valid bearer credential for cache requests from diff --git a/act/common/git/git.go b/act/common/git/git.go index 0ef1eab5..fab65532 100644 --- a/act/common/git/git.go +++ b/act/common/git/git.go @@ -16,6 +16,7 @@ import ( "sync" "gitea.com/gitea/runner/act/common" + "gitea.com/gitea/runner/internal/pkg/lock" "github.com/go-git/go-git/v5" "github.com/go-git/go-git/v5/config" @@ -32,7 +33,7 @@ var ( githubHTTPRegex = regexp.MustCompile(`^https?://.*github.com.*/(.+)/(.+?)(?:.git)?$`) githubSSHRegex = regexp.MustCompile(`github.com[:/](.+)/(.+?)(?:.git)?$`) - cloneLocks sync.Map // key: clone target directory; value: *sync.Mutex + cloneLocks lock.Keyed[string] // key: clone target directory ErrShortRef = errors.New("short SHA references are not supported") ErrNoRepo = errors.New("unable to find git repo") @@ -43,10 +44,7 @@ var ( // Callers reading files inside dir (e.g. tarring a checked-out action into a job container) must hold this lock too, // otherwise a concurrent NewGitCloneExecutor on the same dir can mutate the worktree mid-read. func AcquireCloneLock(dir string) func() { - v, _ := cloneLocks.LoadOrStore(dir, &sync.Mutex{}) - mu := v.(*sync.Mutex) - mu.Lock() - return mu.Unlock + return cloneLocks.Lock(dir) } type Error struct { diff --git a/act/common/git/git_test.go b/act/common/git/git_test.go index 02811acd..6984130e 100644 --- a/act/common/git/git_test.go +++ b/act/common/git/git_test.go @@ -12,7 +12,6 @@ import ( "path/filepath" "strconv" "strings" - "sync" "syscall" "testing" "time" @@ -610,12 +609,4 @@ func TestAcquireCloneLock(t *testing.T) { t.Fatal("acquire on a different directory must not block") } }) - - t.Run("same directory reuses the same mutex", func(t *testing.T) { - dir := t.TempDir() - - v1, _ := cloneLocks.LoadOrStore(dir, &sync.Mutex{}) - v2, _ := cloneLocks.LoadOrStore(dir, &sync.Mutex{}) - require.Same(t, v1, v2) - }) } diff --git a/act/common/outbound_ip.go b/act/common/outbound_ip.go index 2dbed46f..974b1ccc 100644 --- a/act/common/outbound_ip.go +++ b/act/common/outbound_ip.go @@ -19,7 +19,9 @@ func GetOutboundIP() net.IP { conn, err := net.Dial("udp", "8.8.8.8:80") if err == nil { defer conn.Close() - return conn.LocalAddr().(*net.UDPAddr).IP + if addr, ok := conn.LocalAddr().(*net.UDPAddr); ok { + return addr.IP + } } // So the machine cannot access the internet. Pick an IP address from network interfaces. diff --git a/act/container/docker_create_flags_test.go b/act/container/docker_create_flags_test.go index ddbe3814..705c3b14 100644 --- a/act/container/docker_create_flags_test.go +++ b/act/container/docker_create_flags_test.go @@ -51,7 +51,8 @@ func TestCreateFlagsValidate(t *testing.T) { func TestNewContainerAppliesCreateFlags(t *testing.T) { input := &NewContainerInput{Platform: "linux/amd64", Options: "--platform linux/arm64 --pull never"} - cr := NewContainer(input).(*containerReference) + cr, ok := NewContainer(input).(*containerReference) + require.True(t, ok) assert.Equal(t, "linux/arm64", input.Platform) assert.Equal(t, pullPolicyNever, cr.pullPolicy) diff --git a/act/exprparser/interpreter_test.go b/act/exprparser/interpreter_test.go index 8981ed3a..b1484e11 100644 --- a/act/exprparser/interpreter_test.go +++ b/act/exprparser/interpreter_test.go @@ -11,6 +11,7 @@ import ( "gitea.com/gitea/runner/act/model" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestLiterals(t *testing.T) { @@ -523,7 +524,9 @@ func TestOperatorsBooleanEvaluation(t *testing.T) { assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act if expected, ok := tt.expected.(float64); ok && math.IsNaN(expected) { - assert.True(t, math.IsNaN(output.(float64))) + number, ok := output.(float64) + require.True(t, ok, "want a number, got %T", output) + assert.True(t, math.IsNaN(number)) } else { assert.Equal(t, tt.expected, output) } diff --git a/act/model/workflow.go b/act/model/workflow.go index 01ba049f..0b4370b4 100644 --- a/act/model/workflow.go +++ b/act/model/workflow.go @@ -86,11 +86,12 @@ func (w *Workflow) OnSchedule() []string { case []any: allSchedules := []string{} for _, v := range val { - for k, cron := range v.(map[string]any) { - if k != "cron" { - continue - } - allSchedules = append(allSchedules, cron.(string)) + entry, ok := v.(map[string]any) + if !ok { + continue + } + if cron, ok := entry["cron"].(string); ok { + allSchedules = append(allSchedules, cron) } } return allSchedules @@ -443,9 +444,9 @@ func normalizeMatrixValue(key string, val any) ([]any, error) { // Scalar values are wrapped into single-element arrays automatically. // Template expressions are resolved by EvaluateYamlNode before this method is // called; if unresolved, the literal string is wrapped as a one-element fallback. -func (j *Job) Matrix() map[string][]any { +func (j *Job) Matrix() (map[string][]any, error) { if j.Strategy == nil || j.Strategy.RawMatrix.Kind != yaml.MappingNode { - return nil + return map[string][]any{}, nil } // Decode to flexible map first so that scalar values don't cause a type error. @@ -455,9 +456,9 @@ func (j *Job) Matrix() map[string][]any { // Fall back to the strict array-only format for backward compatibility. var val map[string][]any if !decodeNode(j.Strategy.RawMatrix, &val) { - return nil + return map[string][]any{}, nil } - return val + return val, nil } // Convert flexible format to expected format with validation @@ -465,12 +466,11 @@ func (j *Job) Matrix() map[string][]any { for k, v := range flexVal { normalized, err := normalizeMatrixValue(k, v) if err != nil { - log.Errorf("matrix validation error: %v", err) - return nil + return nil, err } val[k] = normalized } - return val + return val, nil } // GetMatrixes returns the matrix cross product @@ -482,38 +482,38 @@ func (j *Job) GetMatrixes() ([]map[string]any, error) { j.Strategy.FailFast = j.Strategy.GetFailFast() j.Strategy.MaxParallel = j.Strategy.GetMaxParallel() - if m := j.Matrix(); m != nil { + m, err := j.Matrix() + if err != nil { + return nil, err + } + if len(m) > 0 { includes := make([]map[string]any, 0) extraIncludes := make([]map[string]any, 0) + addInclude := func(raw any) error { + include, ok := raw.(map[string]any) + if !ok { + return fmt.Errorf("the workflow is not valid. Matrix include %v is not a map of matrix keys to values", raw) + } + for k := range include { + if _, ok := m[k]; ok { + includes = append(includes, include) + return nil + } + } + extraIncludes = append(extraIncludes, include) + return nil + } for _, v := range m["include"] { switch t := v.(type) { case []any: for _, i := range t { - i := i.(map[string]any) - extraInclude := true - for k := range i { - if _, ok := m[k]; ok { - includes = append(includes, i) - extraInclude = false - break - } - } - if extraInclude { - extraIncludes = append(extraIncludes, i) + if err := addInclude(i); err != nil { + return nil, err } } case any: - v := v.(map[string]any) - extraInclude := true - for k := range v { - if _, ok := m[k]; ok { - includes = append(includes, v) - extraInclude = false - break - } - } - if extraInclude { - extraIncludes = append(extraIncludes, v) + if err := addInclude(t); err != nil { + return nil, err } } } @@ -521,10 +521,13 @@ func (j *Job) GetMatrixes() ([]map[string]any, error) { excludes := make([]map[string]any, 0) for _, e := range m["exclude"] { - e := e.(map[string]any) - for k := range e { + exclude, ok := e.(map[string]any) + if !ok { + return nil, fmt.Errorf("the workflow is not valid. Matrix exclude %v is not a map of matrix keys to values", e) + } + for k := range exclude { if _, ok := m[k]; ok { - excludes = append(excludes, e) + excludes = append(excludes, exclude) } else { // We fail completely here because that's what GitHub does for non-existing matrix keys, fail on exclude, silent skip on include return nil, fmt.Errorf("the workflow is not valid. Matrix exclude key %q does not match any key within the matrix", k) diff --git a/act/model/workflow_test.go b/act/model/workflow_test.go index 9f3b5bda..aa6be1d0 100644 --- a/act/model/workflow_test.go +++ b/act/model/workflow_test.go @@ -667,7 +667,9 @@ func TestReadWorkflow_Strategy(t *testing.T) { matrixes, err := job.GetMatrixes() assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act assert.Equal(t, matrixes, []map[string]any{{}}) //nolint:testifylint // pre-existing issue from nektos/act - assert.Equal(t, job.Matrix(), map[string][]any(nil)) + matrix, err := job.Matrix() + require.NoError(t, err) + assert.Empty(t, matrix) assert.Equal(t, job.Strategy.MaxParallel, 2) //nolint:testifylint // pre-existing issue from nektos/act assert.Equal(t, job.Strategy.FailFast, true) //nolint:testifylint // pre-existing issue from nektos/act @@ -675,7 +677,9 @@ func TestReadWorkflow_Strategy(t *testing.T) { matrixes, err = job.GetMatrixes() assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act assert.Equal(t, matrixes, []map[string]any{{}}) //nolint:testifylint // pre-existing issue from nektos/act - assert.Equal(t, job.Matrix(), map[string][]any(nil)) + matrix, err = job.Matrix() + require.NoError(t, err) + assert.Empty(t, matrix) assert.Equal(t, job.Strategy.MaxParallel, 4) //nolint:testifylint // pre-existing issue from nektos/act assert.Equal(t, job.Strategy.FailFast, false) //nolint:testifylint // pre-existing issue from nektos/act @@ -683,7 +687,9 @@ func TestReadWorkflow_Strategy(t *testing.T) { matrixes, err = job.GetMatrixes() assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act assert.Equal(t, matrixes, []map[string]any{{}}) //nolint:testifylint // pre-existing issue from nektos/act - assert.Equal(t, job.Matrix(), map[string][]any(nil)) + matrix, err = job.Matrix() + require.NoError(t, err) + assert.Empty(t, matrix) assert.Equal(t, job.Strategy.MaxParallel, 2) //nolint:testifylint // pre-existing issue from nektos/act assert.Equal(t, job.Strategy.FailFast, false) //nolint:testifylint // pre-existing issue from nektos/act @@ -700,7 +706,9 @@ func TestReadWorkflow_Strategy(t *testing.T) { {"datacenter": "site-b", "node-version": "12.x", "site": "dev"}, }, ) - assert.Equal(t, job.Matrix(), //nolint:testifylint // pre-existing issue from nektos/act + matrix, err = job.Matrix() + require.NoError(t, err) + assert.Equal(t, matrix, //nolint:testifylint // pre-existing issue from nektos/act map[string][]any{ "datacenter": {"site-c", "site-d"}, "exclude": { @@ -1092,13 +1100,15 @@ jobs: t.Fatal("job not found") } - matrix := job.Matrix() + matrix, err := job.Matrix() if tt.wantErr { + require.Error(t, err) assert.Nil(t, matrix, "matrix should be nil on error") } else { + require.NoError(t, err) if tt.wantLen == 0 { - assert.Nil(t, matrix, "matrix should be nil for jobs without strategy") + assert.Empty(t, matrix, "no matrix for jobs without strategy") } else { assert.NotNil(t, matrix, "matrix should not be nil") assert.Len(t, matrix, tt.wantLen, "matrix should have expected number of keys") @@ -1130,11 +1140,9 @@ func TestJobMatrixValidation(t *testing.T) { }, } - // Attempt to get matrix - matrix := job.Matrix() - - // Should return nil due to validation error - assert.Nil(t, matrix, "matrix with nested map should return nil") + matrix, err := job.Matrix() + require.ErrorContains(t, err, `matrix key "config" has invalid nested object value`) + assert.Nil(t, matrix) }) } diff --git a/act/runner/action.go b/act/runner/action.go index c8f7a141..7f2712c6 100644 --- a/act/runner/action.go +++ b/act/runner/action.go @@ -129,6 +129,16 @@ func readActionImpl(ctx context.Context, step *model.Step, actionDir, actionPath return action, err } +// cachedActionTar returns the action's tree from the action cache, which only a remote action +// has an entry in. +func cachedActionTar(ctx context.Context, step actionStep, name, includePrefix string) (io.ReadCloser, error) { + remote, ok := step.(*stepActionRemote) + if !ok { + return nil, fmt.Errorf("action %q is a remote action but runs as %T", name, step) + } + return step.getRunContext().Config.ActionCache.GetTarArchive(ctx, remote.cacheDir, remote.resolvedSha, includePrefix) +} + func maybeCopyToActionDir(ctx context.Context, step actionStep, actionDir, actionPath, containerActionDir string) error { logger := common.Logger(ctx) rc := step.getRunContext() @@ -147,8 +157,7 @@ func maybeCopyToActionDir(ctx context.Context, step actionStep, actionDir, actio } if rc.Config != nil && rc.Config.ActionCache != nil { - raction := step.(*stepActionRemote) - ta, err := rc.Config.ActionCache.GetTarArchive(ctx, raction.cacheDir, raction.resolvedSha, "") + ta, err := cachedActionTar(ctx, step, stepModel.Uses, "") if err != nil { return err } @@ -351,8 +360,7 @@ func execAsDocker(ctx context.Context, step actionStep, actionName, actionDir, b } defer buildContext.Close() } else if rc.Config.ActionCache != nil { - rstep := step.(*stepActionRemote) - buildContext, err = rc.Config.ActionCache.GetTarArchive(ctx, rstep.cacheDir, rstep.resolvedSha, contextDir) + buildContext, err = cachedActionTar(ctx, step, actionName, contextDir) if err != nil { return err } diff --git a/act/runner/run_context.go b/act/runner/run_context.go index 777c51af..5b0a1b21 100644 --- a/act/runner/run_context.go +++ b/act/runner/run_context.go @@ -22,13 +22,13 @@ import ( "runtime" "slices" "strings" - "sync" "time" "gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/container" "gitea.com/gitea/runner/act/exprparser" "gitea.com/gitea/runner/act/model" + "gitea.com/gitea/runner/internal/pkg/lock" "github.com/docker/cli/cli/compose/loader" "github.com/docker/go-connections/nat" @@ -715,13 +715,10 @@ func (rc *RunContext) ActionCacheDir() string { // jobMutexes serializes per-job result/output aggregation across the matrix combinations that // share one *model.Job and run in parallel. Keyed by the shared *model.Job (mirrors the // per-directory AcquireCloneLock pattern). -var jobMutexes sync.Map // key: *model.Job; value: *sync.Mutex +var jobMutexes lock.Keyed[*model.Job] func lockJob(job *model.Job) func() { - v, _ := jobMutexes.LoadOrStore(job, &sync.Mutex{}) - mu := v.(*sync.Mutex) - mu.Lock() - return mu.Unlock + return jobMutexes.Lock(job) } func (rc *RunContext) interpolateOutputs() common.Executor { diff --git a/act/runner/step_action_remote.go b/act/runner/step_action_remote.go index 0cccb729..9d038bd6 100644 --- a/act/runner/step_action_remote.go +++ b/act/runner/step_action_remote.go @@ -138,9 +138,10 @@ func (sar *stepActionRemote) prepareActionExecutor() common.Executor { }) var ntErr common.Executor if err := gitClone(ctx); err != nil { - if errors.Is(err, git.ErrShortRef) { + var refErr *git.Error + if errors.As(err, &refErr) && errors.Is(err, git.ErrShortRef) { return fmt.Errorf("Unable to resolve action `%s`, the provided ref `%s` is the shortened version of a commit SHA, which is not supported. Please use the full commit SHA `%s` instead", - sar.Step.Uses, sar.remoteAction.Ref, err.(*git.Error).Commit()) + sar.Step.Uses, sar.remoteAction.Ref, refErr.Commit()) } else if errors.Is(err, gogit.ErrForceNeeded) { // TODO: figure out if it will be easy to shadow/alias go-git err's ntErr = common.NewInfoExecutor("Non-terminating error while running 'git clone': %v", err) } else { diff --git a/internal/pkg/lock/keyed.go b/internal/pkg/lock/keyed.go new file mode 100644 index 00000000..ee176569 --- /dev/null +++ b/internal/pkg/lock/keyed.go @@ -0,0 +1,38 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package lock + +import "sync" + +// Keyed serializes the work done under one key while letting different keys run in +// parallel. Its zero value is ready to use, and it keeps one mutex per key it has seen. +type Keyed[K comparable] struct { + mu sync.Mutex + mutexes map[K]*sync.Mutex +} + +// Lock locks the mutex for key and returns its unlock function. +func (kl *Keyed[K]) Lock(key K) func() { + kl.mu.Lock() + if kl.mutexes == nil { + kl.mutexes = map[K]*sync.Mutex{} + } + lock := kl.mutexes[key] + if lock == nil { + lock = &sync.Mutex{} + kl.mutexes[key] = lock + } + kl.mu.Unlock() + + lock.Lock() + return lock.Unlock +} + +// Delete drops the mutex for key. Holders of it keep working, they just no longer share it +// with a later Lock of the same key. +func (kl *Keyed[K]) Delete(key K) { + kl.mu.Lock() + delete(kl.mutexes, key) + kl.mu.Unlock() +} diff --git a/internal/pkg/report/reporter.go b/internal/pkg/report/reporter.go index 72cc2b4f..bb82c89d 100644 --- a/internal/pkg/report/reporter.go +++ b/internal/pkg/report/reporter.go @@ -32,6 +32,12 @@ const ( maxOutputValueLen = 1024 * 1024 // 1 MiB ) +// jobOutput is a job output on its way to the server, sent once the server has acknowledged it. +type jobOutput struct { + value string + sent bool +} + type Reporter struct { ctx context.Context cancel context.CancelFunc @@ -53,7 +59,8 @@ type Reporter struct { state *runnerv1.TaskState stateChanged bool stateMu sync.RWMutex - outputs sync.Map + outputsMu sync.Mutex + outputs map[string]jobOutput daemon chan struct{} heartbeatStop chan struct{} heartbeatStopOnce sync.Once @@ -394,7 +401,12 @@ func (r *Reporter) logf(format string, a ...any) { func (r *Reporter) SetOutputs(outputs map[string]string) { r.stateMu.Lock() defer r.stateMu.Unlock() + r.outputsMu.Lock() + defer r.outputsMu.Unlock() + if r.outputs == nil { + r.outputs = map[string]jobOutput{} + } for k, v := range outputs { if l := len(k); l > maxOutputKeyLen { log.Warnf("ignore output %q because the key is too long: %d > %d", k, l, maxOutputKeyLen) @@ -406,10 +418,9 @@ func (r *Reporter) SetOutputs(outputs map[string]string) { r.logf("ignore output %q because the value is too long: %d > %d", k, l, maxOutputValueLen) continue } - if _, ok := r.outputs.Load(k); ok { - continue + if _, ok := r.outputs[k]; !ok { + r.outputs[k] = jobOutput{value: v} } - r.outputs.Store(k, v) } } @@ -574,14 +585,14 @@ func (r *Reporter) ReportState(reportResult bool) error { r.clientM.Lock() defer r.clientM.Unlock() - // Build the outputs map first (single Range pass instead of two). outputs := make(map[string]string) - r.outputs.Range(func(k, v any) bool { - if val, ok := v.(string); ok { - outputs[k.(string)] = val + r.outputsMu.Lock() + for key, out := range r.outputs { + if !out.sent { + outputs[key] = out.value } - return true - }) + } + r.outputsMu.Unlock() // Consume stateChanged atomically with the snapshot; restored on error // below so a concurrent Fire() during UpdateTask isn't silently lost. @@ -594,7 +605,8 @@ func (r *Reporter) ReportState(reportResult bool) error { r.stateMu.Unlock() return nil } - state := proto.Clone(r.state).(*runnerv1.TaskState) + state := &runnerv1.TaskState{} + proto.Merge(state, r.state) r.stateChanged = false r.stateMu.Unlock() @@ -622,21 +634,23 @@ func (r *Reporter) ReportState(reportResult bool) error { metrics.ReportStateTotal.WithLabelValues(metrics.LabelResultSuccess).Inc() r.lastReportedAtNanos.Store(time.Now().UnixNano()) + var noSent []string + r.outputsMu.Lock() for _, k := range resp.Msg.SentOutputs { - r.outputs.Store(k, struct{}{}) + if _, ok := r.outputs[k]; ok { + r.outputs[k] = jobOutput{sent: true} + } } + for key, out := range r.outputs { + if !out.sent { + noSent = append(noSent, key) + } + } + r.outputsMu.Unlock() if resp.Msg.State != nil && resp.Msg.State.Result == runnerv1.Result_RESULT_CANCELLED { r.cancel() } - - var noSent []string - r.outputs.Range(func(k, v any) bool { - if _, ok := v.(string); ok { - noSent = append(noSent, k.(string)) - } - return true - }) if len(noSent) > 0 { return fmt.Errorf("there are still outputs that have not been sent: %v", noSent) } diff --git a/internal/pkg/report/reporter_test.go b/internal/pkg/report/reporter_test.go index 06be9307..0609ab3f 100644 --- a/internal/pkg/report/reporter_test.go +++ b/internal/pkg/report/reporter_test.go @@ -7,6 +7,8 @@ import ( "context" "errors" "fmt" + "maps" + "slices" "strings" "sync/atomic" "testing" @@ -1011,33 +1013,59 @@ func TestReporter_SetOutputs(t *testing.T) { r := &Reporter{state: &runnerv1.TaskState{}} r.SetOutputs(map[string]string{"foo": "bar"}) - got, ok := r.outputs.Load("foo") + got, ok := r.outputs["foo"] require.True(t, ok) - assert.Equal(t, "bar", got) + assert.Equal(t, "bar", got.value) // first value wins: a later write to the same key is ignored r.SetOutputs(map[string]string{"foo": "baz"}) - got, _ = r.outputs.Load("foo") - assert.Equal(t, "bar", got) + assert.Equal(t, "bar", r.outputs["foo"].value) // keys longer than maxOutputKeyLen are dropped longKey := strings.Repeat("k", maxOutputKeyLen+1) r.SetOutputs(map[string]string{longKey: "v"}) - _, ok = r.outputs.Load(longKey) + _, ok = r.outputs[longKey] assert.False(t, ok) // values longer than maxOutputValueLen are dropped longValue := strings.Repeat("v", maxOutputValueLen+1) r.SetOutputs(map[string]string{"big": longValue}) - _, ok = r.outputs.Load("big") + _, ok = r.outputs["big"] assert.False(t, ok) // a value at exactly the limit is still stored maxValue := strings.Repeat("v", maxOutputValueLen) r.SetOutputs(map[string]string{"atlimit": maxValue}) - got, ok = r.outputs.Load("atlimit") + got, ok = r.outputs["atlimit"] require.True(t, ok) - assert.Len(t, got, maxOutputValueLen) + assert.Len(t, got.value, maxOutputValueLen) +} + +// An output the server acknowledged is not reported again. +func TestReporter_OutputsSentOnce(t *testing.T) { + client := mocks.NewClient(t) + var reported []map[string]string + client.On("UpdateTask", mock.Anything, mock.Anything).Return( + func(_ context.Context, req *connect_go.Request[runnerv1.UpdateTaskRequest]) (*connect_go.Response[runnerv1.UpdateTaskResponse], error) { + reported = append(reported, req.Msg.Outputs) + return connect_go.NewResponse(&runnerv1.UpdateTaskResponse{SentOutputs: slices.Collect(maps.Keys(req.Msg.Outputs))}), nil + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + taskCtx, err := structpb.NewStruct(map[string]any{}) + require.NoError(t, err) + cfg, _ := config.LoadDefault("") + r := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg) + + r.SetOutputs(map[string]string{"foo": "bar"}) + require.NoError(t, r.ReportState(false)) + assert.True(t, r.outputs["foo"].sent) + + require.NoError(t, r.ReportState(true)) + require.Len(t, reported, 2) + assert.Equal(t, map[string]string{"foo": "bar"}, reported[0]) + assert.Empty(t, reported[1]) } func TestReporter_EffectiveCloseTimeout(t *testing.T) {