nprasad2077/runner
1
0 Fork 0
mirror of https://gitea.com/gitea/runner.git synced 2026-11-09 16:50:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: remove dead code and consolidate tests (#1335)

Browse Source 
Removes dead code and redundant logic and consolidates tests, so there is less to read and maintain. No behavior change.

1. **act/runner**: inline single-use wrappers, compile regexes once instead of per call, embed the trampoline as a string, drop the unused legacy result of `parseActionCommand`
2. **act/container**: `sanitizeConfig` and `mergeContainerConfigs` stop returning the configs they already mutate in place, drop never-set option fields and the unused `user` parameter of `ExecWithCmdLine`
3. **act/common and cache server**: drop the unused `Trace`/`Tracef` loggers and the short SHA of `FindGitRevision`, `RegisterJob` releases through `RevokeJob`
4. **daemon, poller and reporter**: one shutdown wait for both poll modes, drop an unreachable fetch timeout branch, guard the reporter's last report time with its existing lock instead of an atomic
5. **build**: drop unused Makefile and release workflow variables, collapse the goreleaser ignore list to the same build targets
6. **tests**: fold near-duplicate tests into tables with shared setup

Throughout, hand-rolled logic gives way to `cmp.Or`, `strings.Cut`, `slices` and `min`/`max`.

Reviewed-on: https://gitea.com/gitea/runner/pulls/1335
Reviewed-by: bircni <[email protected]>
Co-authored-by: silverwind <[email protected]>
This commit is contained in:
silverwind authored and silverwind committed 2026-10-22 19:41:10 +00:00
1 parent 923a0bfb59
commit d882311ec6
107 files changed
+2408 -6500
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/pkg/action/checkout/checkout.go
+1 -5
View File
@@ -56,7 +56,7 @@ func Main(ctx context.Context, c *action.Context) (err error) {
ref := c.Inputs["ref"]
sshKey := strings.TrimSpace(c.Inputs["ssh-key"])
isWorkflowRepo := strings.EqualFold(repository, ghc.Repository)
local := isWorkflowRepo && slices.Contains([]string{"", shortBranch(ghc.Ref), ghc.Sha}, shortBranch(ref)) && !c.NoSkipCheckout && sshKey == ""
local := isWorkflowRepo && slices.Contains([]string{"", strings.TrimPrefix(ghc.Ref, "refs/heads/"), ghc.Sha}, strings.TrimPrefix(ref, "refs/heads/")) && !c.NoSkipCheckout && sshKey == ""
if local && (!c.BindWorkdir || dst == c.Workspace) {
if !c.BindWorkdir {
if err := c.Container.CopyDir(dst, c.HostWorkdir+string(filepath.Separator)+".", c.UseGitIgnore, false)(ctx); err != nil {
@@ -470,7 +470,3 @@ func fetchDepth(input string) (int, error) {
}
return depth, nil
}
func shortBranch(ref string) string { // full tag refs stay exact
return strings.TrimPrefix(ref, "refs/heads/")
}
internal/pkg/config/config.go
+5 -13
View File
@@ -246,7 +246,7 @@ type Config struct {
func LoadDefault(file string) (*Config, error) {
// Seeded before the file is read, so a written 0 can mean off.
cfg := &Config{Cache: DefaultCache(), Log: Log{Job: LogJob{Retention: 7 * 24 * time.Hour, MaxSize: 1024 * 1024 * 1024}}}
definedRunnerKeys := map[string]bool{}
var definedRunnerKeys map[string]bool
if file != "" {
content, err := os.ReadFile(file)
if err != nil {
@@ -307,16 +307,13 @@ func LoadDefault(file string) (*Config, error) {
cfg.Runner.Timeout = 3 * time.Hour
}
if cfg.Runner.ActionShallowClone == nil {
b := true
cfg.Runner.ActionShallowClone = &b
cfg.Runner.ActionShallowClone = new(true)
}
if cfg.Runner.SetActEnv == nil {
b := true
cfg.Runner.SetActEnv = &b
cfg.Runner.SetActEnv = new(true)
}
if cfg.Cache.Enabled == nil {
b := true
cfg.Cache.Enabled = &b
cfg.Cache.Enabled = new(true)
}
// Resolved regardless of cache.enabled, because the `cache-server` command reads the secret from the same key without checking cache.enabled.
if err := resolveSecretFile("cache.external_secret", &cfg.Cache.ExternalSecret, cfg.Cache.ExternalSecretFile); err != nil {
@@ -435,12 +432,7 @@ func LoadDefault(file string) (*Config, error) {
// although `container.network_mode` will be deprecated, but we have to be compatible with it for now.
if cfg.Container.NetworkMode != "" && cfg.Container.Network == "" {
logging.Warn("You are trying to use deprecated configuration item of `container.network_mode`, please use `container.network` instead.")
if cfg.Container.NetworkMode == "bridge" {
// Previously, if the value of `container.network_mode` is `bridge`, we will create a new network for job.
// But “bridge” is easily confused with the bridge network created by Docker by default.
// So we set the value of `container.network` to empty string to make `runner` automatically create a new network for job.
cfg.Container.Network = ""
} else {
if cfg.Container.NetworkMode != "bridge" { // "bridge" meant a per-job network, which an empty container.network creates
cfg.Container.Network = cfg.Container.NetworkMode
}
}
internal/pkg/config/config_test.go
+104 -284
View File
@@ -13,6 +13,7 @@ import (
"math/big"
"os"
"path/filepath"
"strings"
"testing"
"time"
@@ -22,34 +23,6 @@ import (
"github.com/stretchr/testify/require"
)
func TestLoadDefault_RejectsExternalServerWithoutSecret(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
cache:
enabled: true
external_server: "http://cache.invalid/"
`), 0o600))
_, err := LoadDefault(path)
require.Error(t, err)
assert.Contains(t, err.Error(), "external_secret")
}
func TestLoadDefault_AcceptsExternalServerWithSecret(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
cache:
enabled: true
external_server: "http://cache.invalid/"
external_secret: "shh"
`), 0o600))
_, err := LoadDefault(path)
require.NoError(t, err)
}
func TestClientCertIsValidatedAndAbsoluteOnLoadReadOnUseAndOfferedOnlyWhenAccepted(t *testing.T) {
require.True(t, (&Runner{Insecure: true}).TLSConfig().InsecureSkipVerify)
@@ -156,13 +129,8 @@ cache:
}
func TestLoadDefault_ToolCacheMode(t *testing.T) {
cfg, err := LoadDefault("")
require.NoError(t, err)
assert.Equal(t, ToolCacheModeNone, cfg.Runner.ToolCacheMode)
path := filepath.Join(t.TempDir(), "config.yaml")
require.NoError(t, os.WriteFile(path, []byte("runner:\n tool_cache_mode: shared\n"), 0o600))
cfg, err = LoadDefault(path)
path := write(t, "runner:\n tool_cache_mode: shared\n")
cfg, err := LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, ToolCacheModeShared, cfg.Runner.ToolCacheMode)
@@ -178,9 +146,7 @@ func TestLoadDefault_RejectsInvalidLogReportLevel(t *testing.T) {
}
func TestLoadDefault_LoadsExtraHeaders(t *testing.T) {
path := filepath.Join(t.TempDir(), "config.yaml")
require.NoError(t, os.WriteFile(path, []byte("runner:\n extra_headers:\n X-Proxy-Token: proxy-token\n"), 0o600))
path := write(t, "runner:\n extra_headers:\n X-Proxy-Token: proxy-token\n")
cfg, err := LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, map[string]string{"X-Proxy-Token": "proxy-token"}, cfg.Runner.ExtraHeaders)
@@ -196,17 +162,13 @@ func TestLoadDefault_LoadsExtraHeaders(t *testing.T) {
}
}
func TestLoadDefault_DefaultsWorkdirCleanupAge(t *testing.T) {
func TestLoadDefault_Defaults(t *testing.T) {
cfg, err := LoadDefault("")
require.NoError(t, err)
assert.Equal(t, 24*time.Hour, cfg.Runner.WorkdirCleanupAge)
assert.Equal(t, 10*time.Minute, cfg.Runner.IdleCleanupInterval)
assert.Equal(t, DefaultImage, cfg.Runner.DefaultImage)
}
func TestLoadDefault_HealthChecksAreOptIn(t *testing.T) {
cfg, err := LoadDefault("")
require.NoError(t, err)
assert.Equal(t, ToolCacheModeNone, cfg.Runner.ToolCacheMode)
assert.False(t, cfg.HealthCheck.Enabled)
assert.Equal(t, int64(1024), cfg.HealthCheck.MinFreeDiskSpaceMB)
assert.Empty(t, cfg.HealthCheck.Script)
@@ -216,9 +178,7 @@ func TestLoadDefault_HealthChecksAreOptIn(t *testing.T) {
}
func TestLoadDefault_DiskAndReadinessSettings(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
cfg, err := LoadDefault(write(t, `
health_check:
enabled: true
min_free_disk_space_mb: 4096
@@ -227,9 +187,7 @@ health_check:
timeout: 3s
metrics:
readiness_grace: 45s
`), 0o600))
cfg, err := LoadDefault(path)
`))
require.NoError(t, err)
assert.True(t, cfg.HealthCheck.Enabled)
assert.Equal(t, int64(4096), cfg.HealthCheck.MinFreeDiskSpaceMB)
@@ -239,88 +197,40 @@ metrics:
assert.Equal(t, 45*time.Second, cfg.Metrics.ReadinessGrace)
}
func TestLoadDefault_UsesConfiguredWorkdirCleanupAge(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
runner:
workdir_cleanup_age: 2h30m
`), 0o600))
cfg, err := LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, 150*time.Minute, cfg.Runner.WorkdirCleanupAge)
}
func TestLoadDefault_UsesConfiguredIdleCleanupInterval(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
runner:
idle_cleanup_interval: 45m
`), 0o600))
cfg, err := LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, 45*time.Minute, cfg.Runner.IdleCleanupInterval)
}
func TestLoadDefault_AllowsDisablingWorkdirCleanup(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
runner:
workdir_cleanup_age: 0s
idle_cleanup_interval: 0s
`), 0o600))
cfg, err := LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, time.Duration(0), cfg.Runner.WorkdirCleanupAge)
assert.Equal(t, time.Duration(0), cfg.Runner.IdleCleanupInterval)
}
func TestLoadDefault_AllowsNegativeWorkdirCleanupValues(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
runner:
workdir_cleanup_age: -1s
idle_cleanup_interval: -1s
`), 0o600))
cfg, err := LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, -1*time.Second, cfg.Runner.WorkdirCleanupAge)
assert.Equal(t, -1*time.Second, cfg.Runner.IdleCleanupInterval)
func TestLoadDefault_WorkdirCleanup(t *testing.T) {
for _, tc := range []struct {
name, yaml string
age, interval time.Duration
}{
{"configured age", "\nrunner:\n workdir_cleanup_age: 2h30m\n", 150 * time.Minute, 10 * time.Minute},
{"configured interval", "\nrunner:\n idle_cleanup_interval: 45m\n", 24 * time.Hour, 45 * time.Minute},
{"disabled", "\nrunner:\n workdir_cleanup_age: 0s\n idle_cleanup_interval: 0s\n", 0, 0},
{"negative", "\nrunner:\n workdir_cleanup_age: -1s\n idle_cleanup_interval: -1s\n", -time.Second, -time.Second},
} {
t.Run(tc.name, func(t *testing.T) {
cfg, err := LoadDefault(write(t, tc.yaml))
require.NoError(t, err)
assert.Equal(t, tc.age, cfg.Runner.WorkdirCleanupAge)
assert.Equal(t, tc.interval, cfg.Runner.IdleCleanupInterval)
})
}
}
func TestLoadDefault_LoadsPostTaskScript(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
runner:
post_task_script: /usr/local/bin/post-task.sh
post_task_script_timeout: 2m
`), 0o600))
cfg, err := LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, "/usr/local/bin/post-task.sh", cfg.Runner.PostTaskScript)
assert.Equal(t, 2*time.Minute, cfg.Runner.PostTaskScriptTimeout)
}
func TestLoadDefault_DefaultsPostTaskScriptTimeout(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
runner:
post_task_script: /usr/local/bin/post-task.sh
`), 0o600))
cfg, err := LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, 5*time.Minute, cfg.Runner.PostTaskScriptTimeout)
for _, tc := range []struct {
name, timeout string
want time.Duration
}{
{"configured timeout", " post_task_script_timeout: 2m\n", 2 * time.Minute},
{"default timeout", "", 5 * time.Minute},
} {
t.Run(tc.name, func(t *testing.T) {
cfg, err := LoadDefault(write(t, "\nrunner:\n post_task_script: /usr/local/bin/post-task.sh\n"+tc.timeout))
require.NoError(t, err)
assert.Equal(t, "/usr/local/bin/post-task.sh", cfg.Runner.PostTaskScript)
assert.Equal(t, tc.want, cfg.Runner.PostTaskScriptTimeout)
})
}
}
func write(t *testing.T, body string) string {
@@ -363,43 +273,28 @@ func TestLoadDefault_LoadsJobLogs(t *testing.T) {
}
func TestLoadDefault_LoadsJobHooks(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
cfg, err := LoadDefault(write(t, `
runner:
hooks:
job_started: /hooks/started.sh
job_completed: /hooks/completed.sh
`), 0o600))
cfg, err := LoadDefault(path)
`))
require.NoError(t, err)
assert.Equal(t, "/hooks/started.sh", cfg.Runner.Hooks.JobStarted)
assert.Equal(t, "/hooks/completed.sh", cfg.Runner.Hooks.JobCompleted)
}
// TestLoadDefault_MalformedYAMLReturnsParseError pins the error surfaced for
// invalid YAML to the canonical "parse config file" message rather than the
// "for defaults metadata" variant — i.e. the main yaml.Unmarshal runs first.
func TestLoadDefault_MalformedYAMLReturnsParseError(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte("runner:\n capacity: [unterminated\n"), 0o600))
_, err := LoadDefault(path)
_, err := LoadDefault(write(t, "runner:\n capacity: [unterminated\n"))
require.Error(t, err)
assert.Contains(t, err.Error(), "parse config file")
assert.NotContains(t, err.Error(), "defaults metadata")
}
func TestLoadDefault_WarnsOnUnknownKeysButStillLoads(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte("container:\n volumes:\n - /host:/ctr\n privileged: true\n"), 0o600))
hook := logtest.SetDefault(t)
cfg, err := LoadDefault(path)
cfg, err := LoadDefault(write(t, "container:\n volumes:\n - /host:/ctr\n privileged: true\n"))
require.NoError(t, err)
assert.True(t, cfg.Container.Privileged)
require.Len(t, hook.AllRecords(), 1)
@@ -407,142 +302,70 @@ func TestLoadDefault_WarnsOnUnknownKeysButStillLoads(t *testing.T) {
}
func TestContainerNetworkCreateOptions(t *testing.T) {
// Verify that the enable_ipv4/enable_ipv6 YAML keys unmarshal into the *bool fields,
// distinguishing an explicit true/false from an omitted key (nil). A nil here is
// forwarded as-is to Docker, which applies its own default.
loadOptions := func(t *testing.T, yaml string) ContainerNetworkCreateOptions {
t.Helper()
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(yaml), 0o600))
cfg, err := LoadDefault(path)
require.NoError(t, err)
return cfg.Container.NetworkCreateOptions
for _, tc := range []struct {
name, yaml string
want ContainerNetworkCreateOptions
}{
{"enable_ipv6 true unmarshals to non-nil true", "container:\n network_create_options:\n enable_ipv6: true\n", ContainerNetworkCreateOptions{EnableIPv6: new(true)}},
{"enable_ipv6 false unmarshals to non-nil false", "container:\n network_create_options:\n enable_ipv6: false\n", ContainerNetworkCreateOptions{EnableIPv6: new(false)}},
{"enable_ipv4 false unmarshals to non-nil false", "container:\n network_create_options:\n enable_ipv4: false\n", ContainerNetworkCreateOptions{EnableIPv4: new(false)}},
{"omitted keys stay nil", "container:\n network_create_options:\n enable_ipv4: true\n", ContainerNetworkCreateOptions{EnableIPv4: new(true)}},
{"omitted block leaves both nil", "container:\n network: \"\"\n", ContainerNetworkCreateOptions{}},
} {
t.Run(tc.name, func(t *testing.T) {
cfg, err := LoadDefault(write(t, tc.yaml))
require.NoError(t, err)
require.Equal(t, tc.want, cfg.Container.NetworkCreateOptions)
})
}
t.Run("enable_ipv6 true unmarshals to non-nil true", func(t *testing.T) {
opts := loadOptions(t, "container:\n network_create_options:\n enable_ipv6: true\n")
require.NotNil(t, opts.EnableIPv6)
assert.True(t, *opts.EnableIPv6)
})
t.Run("enable_ipv6 false unmarshals to non-nil false", func(t *testing.T) {
opts := loadOptions(t, "container:\n network_create_options:\n enable_ipv6: false\n")
require.NotNil(t, opts.EnableIPv6)
assert.False(t, *opts.EnableIPv6)
})
t.Run("enable_ipv4 false unmarshals to non-nil false", func(t *testing.T) {
opts := loadOptions(t, "container:\n network_create_options:\n enable_ipv4: false\n")
require.NotNil(t, opts.EnableIPv4)
assert.False(t, *opts.EnableIPv4)
})
t.Run("omitted keys stay nil", func(t *testing.T) {
opts := loadOptions(t, "container:\n network_create_options:\n enable_ipv4: true\n")
require.NotNil(t, opts.EnableIPv4)
assert.True(t, *opts.EnableIPv4)
assert.Nil(t, opts.EnableIPv6, "an omitted enable_ipv6 must remain nil so Docker's default applies")
})
t.Run("omitted block leaves both nil", func(t *testing.T) {
opts := loadOptions(t, "container:\n network: \"\"\n")
assert.Nil(t, opts.EnableIPv4)
assert.Nil(t, opts.EnableIPv6)
})
}
func TestLoadDefault_ReadsExternalSecretFromFile(t *testing.T) {
dir := t.TempDir()
secretPath := filepath.Join(dir, "cache.secret")
require.NoError(t, os.WriteFile(secretPath, []byte(" s3cr3t\n"), 0o600))
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
cache:
enabled: true
external_server: "http://cache.invalid/"
external_secret_file: "`+secretPath+`"
`), 0o600))
cfg, err := LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, "s3cr3t", cfg.Cache.ExternalSecret)
func TestLoadDefault_ExternalCacheSecret(t *testing.T) {
const externalCache = "\ncache:\n enabled: true\n external_server: \"http://cache.invalid/\"\n"
for _, tc := range []struct {
name, yaml, fileContent, wantSecret, wantErr string
}{
{name: "without secret", yaml: externalCache, wantErr: "external_secret"},
{name: "inline secret", yaml: externalCache + " external_secret: \"shh\"\n"},
{
name: "from file", yaml: externalCache + " external_secret_file: \"$DIR/cache.secret\"\n",
fileContent: " s3cr3t\n", wantSecret: "s3cr3t",
},
{
name: "from file when disabled", yaml: "\ncache:\n enabled: false\n external_secret_file: \"$DIR/cache.secret\"\n",
fileContent: "s3cr3t", wantSecret: "s3cr3t",
},
{
name: "inline and file conflict", yaml: externalCache + " external_secret: \"inline\"\n external_secret_file: \"$DIR/cache.secret\"\n",
fileContent: "s3cr3t", wantErr: "both set",
},
{name: "missing file", yaml: externalCache + " external_secret_file: \"$DIR/absent.secret\"\n", wantErr: "read cache.external_secret_file"},
{
name: "empty file", yaml: externalCache + " external_secret_file: \"$DIR/cache.secret\"\n",
fileContent: "\n \n", wantErr: "contains no secret",
},
} {
t.Run(tc.name, func(t *testing.T) {
dir := t.TempDir()
if tc.fileContent != "" {
require.NoError(t, os.WriteFile(filepath.Join(dir, "cache.secret"), []byte(tc.fileContent), 0o600))
}
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(strings.ReplaceAll(tc.yaml, "$DIR", dir)), 0o600))
cfg, err := LoadDefault(path)
if tc.wantErr != "" {
require.Error(t, err)
assert.Contains(t, err.Error(), tc.wantErr)
} else {
require.NoError(t, err)
if tc.wantSecret != "" {
assert.Equal(t, tc.wantSecret, cfg.Cache.ExternalSecret)
}
}
})
}
}
func TestLoadDefault_ReadsExternalSecretFromFileWhenCacheDisabled(t *testing.T) {
dir := t.TempDir()
secretPath := filepath.Join(dir, "cache.secret")
require.NoError(t, os.WriteFile(secretPath, []byte("s3cr3t"), 0o600))
// the file has to be resolved even when cache is disabled
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
cache:
enabled: false
external_secret_file: "`+secretPath+`"
`), 0o600))
cfg, err := LoadDefault(path)
require.NoError(t, err)
assert.Equal(t, "s3cr3t", cfg.Cache.ExternalSecret)
}
func TestLoadDefault_RejectsBothExternalSecretAndFile(t *testing.T) {
dir := t.TempDir()
secretPath := filepath.Join(dir, "cache.secret")
require.NoError(t, os.WriteFile(secretPath, []byte("s3cr3t"), 0o600))
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
cache:
enabled: true
external_server: "http://cache.invalid/"
external_secret: "inline"
external_secret_file: "`+secretPath+`"
`), 0o600))
_, err := LoadDefault(path)
require.Error(t, err)
assert.Contains(t, err.Error(), "both set")
}
func TestLoadDefault_RejectsMissingExternalSecretFile(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
cache:
enabled: true
external_server: "http://cache.invalid/"
external_secret_file: "`+filepath.Join(dir, "absent.secret")+`"
`), 0o600))
_, err := LoadDefault(path)
require.Error(t, err)
assert.Contains(t, err.Error(), "read cache.external_secret_file")
}
func TestLoadDefault_RejectsEmptyExternalSecretFile(t *testing.T) {
dir := t.TempDir()
secretPath := filepath.Join(dir, "cache.secret")
require.NoError(t, os.WriteFile(secretPath, []byte("\n \n"), 0o600))
path := filepath.Join(dir, "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
cache:
enabled: true
external_server: "http://cache.invalid/"
external_secret_file: "`+secretPath+`"
`), 0o600))
_, err := LoadDefault(path)
require.Error(t, err)
assert.Contains(t, err.Error(), "contains no secret")
}
// The shipped configs must parse, hold no key the config does not know, and leave
// every option at its default, as all of their values are commented out.
func TestLoadDefault_ShippedConfigsChangeNothing(t *testing.T) {
hook := logtest.SetDefault(t)
@@ -562,13 +385,10 @@ func TestLoadDefault_ShippedConfigsChangeNothing(t *testing.T) {
}
func TestLoadDefault_ClampsFetchTimeout(t *testing.T) {
path := filepath.Join(t.TempDir(), "config.yaml")
require.NoError(t, os.WriteFile(path, []byte(`
cfg, err := LoadDefault(write(t, `
runner:
fetch_timeout: 120s
`), 0o600))
cfg, err := LoadDefault(path)
`))
require.NoError(t, err)
assert.Equal(t, RequestTimeout, cfg.Runner.FetchTimeout)
}
internal/pkg/config/edit.go
+1 -10
View File
@@ -277,7 +277,7 @@ func mappingValue(mapping *yaml.Node, key string) *yaml.Node {
}
func renderNode(node *yaml.Node) (string, error) {
if !allScalars(node.Content) {
if slices.ContainsFunc(node.Content, func(child *yaml.Node) bool { return child.Kind != yaml.ScalarNode }) {
encoded, err := encodeYAML(node)
if err != nil {
return "", err
@@ -303,15 +303,6 @@ func renderNode(node *yaml.Node) (string, error) {
}
}
func allScalars(nodes []*yaml.Node) bool {
for _, node := range nodes {
if node.Kind != yaml.ScalarNode {
return false
}
}
return true
}
// encodeYAML never wraps lines, re-encoding must not reflow long values.
func encodeYAML(node *yaml.Node) ([]byte, error) {
return yaml.Dump(node, yaml.WithV3Defaults(), yaml.WithIndent(2), yaml.WithLineWidth(-1))
internal/pkg/labels/labels.go
+6 -17
View File
@@ -5,6 +5,7 @@ package labels
import (
"errors"
"slices"
"strings"
)
@@ -31,17 +32,10 @@ func Parse(str string) (*Label, error) {
return nil, errors.New("empty label")
}
splits := strings.SplitN(str, ":", 3)
label := &Label{
Name: splits[0],
Schema: SchemeHost,
Arg: "",
}
if len(splits) >= 2 {
label.Schema = splits[1]
}
if len(splits) >= 3 {
label.Arg = splits[2]
name, remainder, hasSchema := strings.Cut(str, ":")
label := &Label{Name: name, Schema: SchemeHost}
if hasSchema {
label.Schema, label.Arg, _ = strings.Cut(remainder, ":")
}
kubernetes := label.Schema == SchemeKubernetes && strings.HasPrefix(label.Arg, "//") && len(label.Arg) > 2
if label.Schema != SchemeHost && label.Schema != SchemeDocker && !kubernetes {
@@ -58,12 +52,7 @@ func Parse(str string) (*Label, error) {
type Labels []*Label
func (l Labels) RequireDocker() bool {
for _, label := range l {
if label.Schema == SchemeDocker {
return true
}
}
return false
return slices.ContainsFunc(l, func(label *Label) bool { return label.Schema == SchemeDocker })
}
// PickPlatform returns the platform of the first runs-on entry this runner has a label for, or "".
internal/pkg/labels/labels_test.go
+9 -58
View File
@@ -11,67 +11,20 @@ import (
func TestParse(t *testing.T) {
tests := []struct {
args string
want *Label
wantErr bool
args string
want *Label
}{
{
args: "ubuntu:docker://node:18",
want: &Label{
Name: "ubuntu",
Schema: "docker",
Arg: "//node:18",
},
wantErr: false,
},
{
args: "ubuntu:host",
want: &Label{
Name: "ubuntu",
Schema: "host",
Arg: "",
},
wantErr: false,
},
{
args: "ubuntu",
want: &Label{
Name: "ubuntu",
Schema: "host",
Arg: "",
},
wantErr: false,
},
{
args: "pool:e57e18d4-10d4-406f-93bf-60f127221bdd",
want: &Label{
Name: "pool:e57e18d4-10d4-406f-93bf-60f127221bdd",
Schema: "host",
Arg: "",
Opaque: true,
},
wantErr: false,
},
{
args: "ubuntu:vm:ubuntu-18.04",
want: &Label{
Name: "ubuntu:vm:ubuntu-18.04",
Schema: "host",
Arg: "",
Opaque: true,
},
wantErr: false,
},
{
args: "",
want: nil,
wantErr: true,
},
{"ubuntu:docker://node:18", &Label{Name: "ubuntu", Schema: "docker", Arg: "//node:18"}},
{"ubuntu:host", &Label{Name: "ubuntu", Schema: "host"}},
{"ubuntu", &Label{Name: "ubuntu", Schema: "host"}},
{"pool:e57e18d4-10d4-406f-93bf-60f127221bdd", &Label{Name: "pool:e57e18d4-10d4-406f-93bf-60f127221bdd", Schema: "host", Opaque: true}},
{"ubuntu:vm:ubuntu-18.04", &Label{Name: "ubuntu:vm:ubuntu-18.04", Schema: "host", Opaque: true}},
{"", nil},
}
for _, tt := range tests {
t.Run(tt.args, func(t *testing.T) {
got, err := Parse(tt.args)
if tt.wantErr {
if tt.want == nil {
require.Error(t, err)
return
}
@@ -81,7 +34,6 @@ func TestParse(t *testing.T) {
}
}
// mustParse parses the given label strings, failing the test on any error.
func mustParse(t *testing.T, strs ...string) Labels {
t.Helper()
ls := make(Labels, 0, len(strs))
@@ -157,7 +109,6 @@ func TestToStrings(t *testing.T) {
}, ls.ToStrings())
}
// a colon-containing name must survive a write to and read back from the .runner file
func TestOpaqueLabelRoundTrip(t *testing.T) {
const raw = "pool:e57e18d4-10d4-406f-93bf-60f127221bdd"
internal/pkg/lock/lock_test.go
+11 -34
View File
@@ -6,49 +6,26 @@
package lock
import (
"errors"
"path/filepath"
"testing"
"github.com/stretchr/testify/require"
)
func TestTryLock(t *testing.T) {
runnerFile := filepath.Join(t.TempDir(), ".runner")
release, err := TryLock(runnerFile)
if err != nil {
t.Fatalf("first TryLock failed: %v", err)
}
// A second lock on the same file must be refused while the first is held.
if _, err := TryLock(runnerFile); !errors.Is(err, ErrLocked) {
t.Fatalf("second TryLock: want ErrLocked, got %v", err)
}
if err := release(); err != nil {
t.Fatalf("release failed: %v", err)
}
// After release the lock is available again.
release2, err := TryLock(runnerFile)
if err != nil {
t.Fatalf("TryLock after release failed: %v", err)
}
if err := release2(); err != nil {
t.Fatalf("second release failed: %v", err)
for range 2 {
release, err := TryLock(runnerFile)
require.NoError(t, err)
_, err = TryLock(runnerFile)
require.ErrorIs(t, err, ErrLocked)
require.NoError(t, release())
}
}
// TestTryLockUncreatable ensures a lock file that cannot be created reports a
// non-ErrLocked error, so callers can tell "already locked by another process"
// apart from "couldn't lock" and degrade gracefully (e.g. a read-only mount).
func TestTryLockUncreatable(t *testing.T) {
runnerFile := filepath.Join(t.TempDir(), "missing-dir", ".runner")
_, err := TryLock(runnerFile)
if err == nil {
t.Fatal("TryLock on an uncreatable lock file: want error, got nil")
}
if errors.Is(err, ErrLocked) {
t.Fatal("TryLock on an uncreatable lock file: want non-ErrLocked error, got ErrLocked")
}
_, err := TryLock(filepath.Join(t.TempDir(), "missing-dir", ".runner"))
require.Error(t, err)
require.NotErrorIs(t, err, ErrLocked)
}
internal/pkg/metrics/server_test.go
+10 -19
View File
@@ -11,33 +11,24 @@ import (
"github.com/stretchr/testify/assert"
)
func TestReadyEndpoint(t *testing.T) {
func TestHealthAndReadyEndpoints(t *testing.T) {
tests := []struct {
name string
ready bool
reason string
status int
name, path string
ready bool
reason, body string
status int
}{
{"ready", true, "ok", http.StatusOK},
{"not ready", false, "low disk space", http.StatusServiceUnavailable},
{"ready", "/readyz", true, "ok", "ok", http.StatusOK},
{"not ready", "/readyz", false, "low disk space", "low disk space", http.StatusServiceUnavailable},
{"live when not ready", "/healthz", false, "low disk space", "ok", http.StatusOK},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
handler := NewHTTPHandler(func() (bool, string) { return test.ready, test.reason })
request := httptest.NewRequest(http.MethodGet, "/readyz", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
handler.ServeHTTP(response, httptest.NewRequest(http.MethodGet, test.path, nil))
assert.Equal(t, test.status, response.Code)
assert.Equal(t, test.reason, response.Body.String())
assert.Equal(t, test.body, response.Body.String())
})
}
}
func TestHealthEndpointStaysLiveWhenNotReady(t *testing.T) {
handler := NewHTTPHandler(func() (bool, string) { return false, "low disk space" })
request := httptest.NewRequest(http.MethodGet, "/healthz", nil)
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
assert.Equal(t, http.StatusOK, response.Code)
assert.Equal(t, "ok", response.Body.String())
}
internal/pkg/process/killer_windows_test.go
+3 -29
View File
@@ -5,11 +5,8 @@ package process
import (
"fmt"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"testing"
"time"
@@ -17,7 +14,6 @@ import (
"golang.org/x/sys/windows"
)
// processAlive reports whether pid refers to a still-running process.
func processAlive(pid int) bool {
h, err := windows.OpenProcess(windows.PROCESS_QUERY_LIMITED_INFORMATION, false, uint32(pid))
if err != nil {
@@ -28,21 +24,13 @@ func processAlive(pid int) bool {
if err := windows.GetExitCodeProcess(h, &code); err != nil {
return false
}
const stillActive = 259 // STILL_ACTIVE
const stillActive = 259
return code == stillActive
}
// TestKillerKillsTree verifies that a process assigned to the Job Object is
// terminated together with a child it spawns afterwards. This mirrors a step or
// post-task script that launches a child which spawns further processes, where
// cancelling must take down the whole tree, not just the direct child.
func TestKillerKillsTree(t *testing.T) {
dir := t.TempDir()
pidFile := filepath.Join(dir, "child.pid")
pidFile := filepath.Join(t.TempDir(), "child.pid")
// Parent powershell spawns a detached, long-lived child powershell (writing
// its PID to a file) and then sleeps. The child is launched AFTER the parent
// has been assigned to the job, so it must be captured by the job too.
script := fmt.Sprintf(
`$c = Start-Process powershell -PassThru -ArgumentList '-NoProfile','-Command','Start-Sleep -Seconds 600'; `+
`Set-Content -LiteralPath %q -Value $c.Id; Start-Sleep -Seconds 600`, pidFile)
@@ -54,22 +42,8 @@ func TestKillerKillsTree(t *testing.T) {
require.NoError(t, err)
defer killer.Close()
// Wait for the child PID to be reported.
var childPID int
require.Eventually(t, func() bool {
b, e := os.ReadFile(pidFile)
if e != nil {
return false
}
s := strings.TrimSpace(string(b))
if s == "" {
return false
}
childPID, _ = strconv.Atoi(s)
return childPID > 0 && processAlive(childPID)
}, 20*time.Second, 200*time.Millisecond, "child process should start")
childPID := awaitChildPID(t, pidFile)
// Killing the job must terminate both the parent and the detached child.
require.NoError(t, killer.Kill())
require.Eventually(t, func() bool {
internal/pkg/process/leftover.go
+5 -13
View File
@@ -8,22 +8,14 @@ import (
"strings"
)
// pathWithin is the OS-agnostic core of dirContainsPath: dir and target must
// already be cleaned, and fold folds case.
func pathWithin(dir, target string, fold bool) bool {
if dir == "" || target == "" || dir == "." || target == "." {
func dirContainsPath(dir, target string) bool {
dir = strings.ToLower(filepath.Clean(dir))
target = strings.ToLower(filepath.Clean(target))
if dir == "." || target == "." {
return false
}
if fold {
dir = strings.ToLower(dir)
target = strings.ToLower(target)
}
if dir == target {
return true
}
sep := string(filepath.Separator)
if !strings.HasSuffix(dir, sep) {
dir += sep
}
return strings.HasPrefix(target, dir)
return strings.HasPrefix(target, strings.TrimSuffix(dir, string(filepath.Separator))+string(filepath.Separator))
}
internal/pkg/process/leftover_test.go
+13 -15
View File
@@ -8,33 +8,31 @@ import (
"testing"
)
func TestPathWithin(t *testing.T) {
func TestDirContainsPath(t *testing.T) {
base := filepath.Join("base", "job1")
cases := []struct {
name string
dir string
target string
fold bool
want bool
}{
{"same dir", base, base, false, true},
{"direct child", base, filepath.Join(base, "app.exe"), false, true},
{"deep child", base, filepath.Join(base, "sub", "sub", "app.exe"), false, true},
{"parent is not within child", filepath.Join(base, "sub"), base, false, false},
{"name-prefix sibling spared", base, filepath.Join("base", "job10", "app.exe"), false, false},
{"unrelated", base, filepath.Join("other", "app.exe"), false, false},
{"empty dir", "", base, false, false},
{"empty target", base, "", false, false},
{"dot dir", ".", base, false, false},
{"case-sensitive miss", base, filepath.Join("base", "JOB1", "app.exe"), false, false},
{"case-insensitive hit", base, filepath.Join("base", "JOB1", "app.exe"), true, true},
{"same dir", base, base, true},
{"direct child", base, filepath.Join(base, "app.exe"), true},
{"deep child", base, filepath.Join(base, "sub", "sub", "app.exe"), true},
{"parent is not within child", filepath.Join(base, "sub"), base, false},
{"name-prefix sibling spared", base, filepath.Join("base", "job10", "app.exe"), false},
{"unrelated", base, filepath.Join("other", "app.exe"), false},
{"empty dir", "", base, false},
{"empty target", base, "", false},
{"dot dir", ".", base, false},
{"case-insensitive hit", base, filepath.Join("base", "JOB1", "app.exe"), true},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := pathWithin(filepath.Clean(tc.dir), filepath.Clean(tc.target), tc.fold)
got := dirContainsPath(tc.dir, tc.target)
if got != tc.want {
t.Fatalf("pathWithin(%q, %q, fold=%v) = %v, want %v", tc.dir, tc.target, tc.fold, got, tc.want)
t.Fatalf("dirContainsPath(%q, %q) = %v, want %v", tc.dir, tc.target, got, tc.want)
}
})
}
internal/pkg/process/leftover_windows.go
-7
View File
@@ -8,18 +8,11 @@ import (
"errors"
"fmt"
"os"
"path/filepath"
"unsafe"
"golang.org/x/sys/windows"
)
// dirContainsPath reports whether target is dir or lives beneath it,
// case-insensitively, so a name-prefix sibling (job1 vs job10) is not a match.
func dirContainsPath(dir, target string) bool {
return pathWithin(filepath.Clean(dir), filepath.Clean(target), true)
}
// KillProcessesWithCWDUnder terminates every process in the runner's session
// whose cwd is one of dirs or below, catching leftovers the path scan misses
// (Win32_Process exposes no cwd) that still pin a handle. Best-effort.
internal/pkg/report/reporter.go
+30 -64
View File
@@ -12,7 +12,6 @@ import (
"slices"
"strings"
"sync"
"sync/atomic"
"time"
"gitea.com/gitea/runner/act/common/logging"
@@ -80,9 +79,7 @@ type Reporter struct {
heartbeatStop chan struct{}
heartbeatStopOnce sync.Once
// Unix-nanos of the last successful UpdateTask. Atomic so the heartbeat
// guard in ReportState reads it without contending stateMu.
lastReportedAtNanos atomic.Int64
lastReportedAtNanos int64 // guarded by clientM
// Adaptive batching control
logReportInterval time.Duration
@@ -110,14 +107,8 @@ func NewReporter(ctx context.Context, cancel context.CancelFunc, client client.C
for _, v := range extraMasks {
oldnew = runner.AppendSecretMasker(oldnew, v)
}
if v := task.Context.Fields["token"].GetStringValue(); v != "" {
oldnew = runner.AppendSecretMasker(oldnew, v)
}
if v := task.Context.Fields["gitea_runtime_token"].GetStringValue(); v != "" {
oldnew = runner.AppendSecretMasker(oldnew, v)
}
if v := task.Context.Fields["actions_id_token_request_token"].GetStringValue(); v != "" {
oldnew = runner.AppendSecretMasker(oldnew, v)
for _, key := range []string{"token", "gitea_runtime_token", "actions_id_token_request_token"} {
oldnew = runner.AppendSecretMasker(oldnew, task.Context.Fields[key].GetStringValue())
}
oldnew = runner.AppendSecretMaskers(oldnew, task.Secrets)
@@ -239,10 +230,7 @@ func (r *Reporter) reportedStepOf(entry logging.Record) (*runnerv1.StepState, bo
default:
return r.currentReportedStep, false
}
main, ok := r.reportedSteps[reportedStepKey{runnerv1.StepStage_STEP_STAGE_MAIN, number}]
if !ok {
return r.currentReportedStep, false
}
main := r.reportedSteps[reportedStepKey{runnerv1.StepStage_STEP_STAGE_MAIN, number}]
if evaluated, _ := recordAttr[string](entry, "step"); evaluated != "" {
// evaluated as GitHub shows it, and masked as an expression can name a secret
main.Name = r.mask(stepDisplayName(r.stepModels[number], evaluated))
@@ -358,16 +346,10 @@ func attrArgs(attrs []logging.Attr) []any {
return args
}
// isJobStepEntry is used to not report composite step results incorrectly as step result
// returns true if the logentry is on job level
// returns false for composite action step messages
// isJobStepEntry is false for composite action steps, whose results are not job step results.
func isJobStepEntry(rec logging.Record) bool {
if v, ok := rec.Attr("stepID"); ok {
if v, ok := v.([]string); ok && len(v) > 1 {
return false
}
}
return true
stepID, _ := recordAttr[[]string](rec, "stepID")
return len(stepID) <= 1
}
// notifyLog sends a non-blocking signal that new log rows are available.
@@ -421,7 +403,7 @@ func (r *Reporter) Handle(_ context.Context, entry logging.Record) error {
urgentState = true
}
if v, ok := entry.Attr("stepResult"); ok && own && isJobStepEntry(entry) {
if stepResult, ok := r.parseResult(v); ok {
if stepResult, ok := parseResult(v); ok {
reported.Result = stepResult
reported.StoppedAt = timestamppb.New(timestamp)
urgentState = true
@@ -433,7 +415,7 @@ func (r *Reporter) Handle(_ context.Context, entry logging.Record) error {
if stage != runner.StageMain {
if v, ok := entry.Attr("jobResult"); ok {
if jobResult, ok := r.parseResult(v); ok {
if jobResult, ok := parseResult(v); ok {
// We need to ensure log upload before this upload
r.state.Result = jobResult
r.state.StoppedAt = timestamppb.New(timestamp)
@@ -476,24 +458,19 @@ func (r *Reporter) Handle(_ context.Context, entry logging.Record) error {
}
// Force reporting log errors as raw output to prevent silent failures
v, ok := entry.Attr("raw_output")
if entry.Level >= logging.LevelError {
v, ok = true, true
}
if ok {
if rawOutput, ok := v.(bool); ok && rawOutput {
if step.LogLength == 0 {
step.LogIndex = int64(r.logOffset + len(r.logRows))
}
step.LogLength++
r.appendLogRow(logRow(entry))
v, hasRawOutput := entry.Attr("raw_output")
rawOutput, _ := v.(bool)
if entry.Level >= logging.LevelError || rawOutput {
if step.LogLength == 0 {
step.LogIndex = int64(r.logOffset + len(r.logRows))
}
} else if r.shouldAppendLogRow(entry) {
step.LogLength++
r.appendLogRow(logRow(entry))
} else if !hasRawOutput && r.shouldAppendLogRow(entry) {
r.appendLogRow(logRow(entry))
}
if v, ok := entry.Attr("stepResult"); ok && isJobStepEntry(entry) {
if stepResult, ok := r.parseResult(v); ok {
if stepResult, ok := parseResult(v); ok {
if step.LogLength == 0 {
step.LogIndex = int64(r.logOffset + len(r.logRows))
}
@@ -535,18 +512,6 @@ func (r *Reporter) StopHeartbeats() {
})
}
func (r *Reporter) stopLatencyTimer(active *bool, timer *time.Timer) {
if *active {
if !timer.Stop() {
select {
case <-timer.C:
default:
}
}
*active = false
}
}
func (r *Reporter) runDaemonLoop() {
defer close(r.daemon)
@@ -556,10 +521,12 @@ func (r *Reporter) runDaemonLoop() {
// maxLatencyTimer ensures the first buffered log row is sent within logReportMaxLatency.
// Start inactive — it is armed when the first log row arrives in an empty buffer.
maxLatencyTimer := time.NewTimer(0)
if !maxLatencyTimer.Stop() {
<-maxLatencyTimer.C
}
maxLatencyTimer.Stop()
maxLatencyActive := false
stopLatencyTimer := func() {
maxLatencyTimer.Stop()
maxLatencyActive = false
}
defer logTicker.Stop()
defer stateTicker.Stop()
@@ -569,7 +536,7 @@ func (r *Reporter) runDaemonLoop() {
select {
case <-logTicker.C:
r.reportLogWithState()
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
stopLatencyTimer()
case <-stateTicker.C:
r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false))
@@ -581,7 +548,7 @@ func (r *Reporter) runDaemonLoop() {
if n >= r.logBatchSize {
r.reportLogWithState()
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
stopLatencyTimer()
} else if !maxLatencyActive && n > 0 {
maxLatencyTimer.Reset(r.logReportMaxLatency)
maxLatencyActive = true
@@ -591,7 +558,7 @@ func (r *Reporter) runDaemonLoop() {
// Step transition or job result — flush both immediately for frontend UX.
r.noteReport(metrics.LabelMethodUpdateLog, r.ReportLog(false))
r.noteReport(metrics.LabelMethodUpdateTask, r.ReportState(false))
r.stopLatencyTimer(&maxLatencyActive, maxLatencyTimer)
stopLatencyTimer()
case <-maxLatencyTimer.C:
maxLatencyActive = false
@@ -882,15 +849,14 @@ func (r *Reporter) ReportState(reportResult bool) error {
// below so a log record handled during UpdateTask isn't silently lost.
// Heartbeat at stateReportInterval even when nothing changed, so the server
// doesn't time out long-running silent jobs as orphaned (#826).
last := r.lastReportedAtNanos.Load()
last := r.lastReportedAtNanos
withinHeartbeatInterval := last != 0 && time.Since(time.Unix(0, last)) < r.stateReportInterval
r.stateMu.Lock()
if !reportResult && !r.stateChanged && len(outputs) == 0 && withinHeartbeatInterval {
r.stateMu.Unlock()
return nil
}
state := &runnerv1.TaskState{}
proto.Merge(state, r.state)
state := proto.CloneOf(r.state)
r.stateChanged = false
r.stateMu.Unlock()
@@ -916,7 +882,7 @@ func (r *Reporter) ReportState(reportResult bool) error {
return err
}
metrics.ReportStateTotal.WithLabelValues(metrics.LabelResultSuccess).Inc()
r.lastReportedAtNanos.Store(time.Now().UnixNano())
r.lastReportedAtNanos = time.Now().UnixNano()
var noSent []string
r.outputsMu.Lock()
@@ -972,7 +938,7 @@ var stringToResult = map[string]runnerv1.Result{
"cancelled": runnerv1.Result_RESULT_CANCELLED,
}
func (r *Reporter) parseResult(result any) (runnerv1.Result, bool) {
func parseResult(result any) (runnerv1.Result, bool) {
str := ""
if v, ok := result.(string); ok { // for jobResult
str = v
internal/pkg/report/reporter_test.go
+67 -312
View File
@@ -50,7 +50,6 @@ func TestReporter_CompositeStepResultDoesNotFinishStep(t *testing.T) {
}
func TestReporter_LogLevelFiltering(t *testing.T) {
// Set global level to Info so Debug entries should be filtered.
origLevel := logging.GetLevel()
logging.SetLevel(logging.LevelInfo)
defer logging.SetLevel(origLevel)
@@ -65,12 +64,9 @@ func TestReporter_LogLevelFiltering(t *testing.T) {
return &runnerv1.UpdateTaskResponse{}, nil
})
ctx, cancel := context.WithCancel(t.Context())
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
cfg, _ := config.LoadDefault("")
cfg.Runner.LogReportLevel = "warn"
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter := newTestReporter(t, client, cfg)
assert.False(t, reporter.shouldAppendLogRow(logging.Record{Level: logging.LevelInfo}))
assert.True(t, reporter.shouldAppendLogRow(logging.Record{Level: logging.LevelWarn}))
reporter.RunDaemon()
@@ -82,16 +78,13 @@ func TestReporter_LogLevelFiltering(t *testing.T) {
dataStep0 := map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true}
dataStep0Internal := map[string]any{"stage": "Main", "stepNumber": 0}
// raw_output entries always appear in job log regardless of level.
require.NoError(t, fire(reporter, logEntry{Message: "step output", Data: dataStep0, Level: logging.LevelInfo}))
require.NoError(t, fire(reporter, logEntry{Message: "step debug output", Data: dataStep0, Level: logging.LevelDebug}))
assert.Equal(t, int64(2), reporter.state.Steps[0].LogLength, "raw_output entries must always be forwarded")
// Non-raw_output entries during steps are not added to logRows regardless of level.
require.NoError(t, fire(reporter, logEntry{Message: "internal info", Data: dataStep0Internal, Level: logging.LevelInfo}))
require.NoError(t, fire(reporter, logEntry{Message: "internal debug", Data: dataStep0Internal, Level: logging.LevelDebug}))
// stepResult at DebugLevel (skipped step) must still update state even when filtered from log.
require.NoError(t, fire(reporter, logEntry{
Message: "Skipping step",
Data: map[string]any{
@@ -105,14 +98,6 @@ func TestReporter_LogLevelFiltering(t *testing.T) {
"stepResult at DebugLevel must update step state even when log entry is filtered from job log output")
}
// TestReporter_EphemeralRunnerDeletion reproduces the exact scenario from
// https://gitea.com/gitea/runner/issues/793:
//
// 1. RunDaemon calls ReportLog(false) — runner is still alive
// 2. Close() updates state to Result=FAILURE (between RunDaemon's ReportLog and ReportState)
// 3. RunDaemon's ReportState() would clone the completed state and send it,
// but the fix makes ReportState return early when closed, preventing this
// 4. Close's ReportLog(true) succeeds because the runner was not deleted
func TestReporter_EphemeralRunnerDeletion(t *testing.T) {
runnerDeleted := false
@@ -129,7 +114,6 @@ func TestReporter_EphemeralRunnerDeletion(t *testing.T) {
)
client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(
func(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
// Server deletes ephemeral runner when it receives a completed state
if req.State != nil && req.State.Result != runnerv1.Result_RESULT_UNSPECIFIED {
runnerDeleted = true
}
@@ -137,25 +121,17 @@ func TestReporter_EphemeralRunnerDeletion(t *testing.T) {
},
)
ctx, cancel := context.WithCancel(t.Context())
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 := newTestReporter(t, client, cfg)
reporter.ResetSteps(newTestSteps(1))
// Fire a log entry to create pending data
require.NoError(t, fire(reporter, logEntry{
Message: "build output",
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
// Step 1: RunDaemon calls ReportLog(false) — runner is still alive
require.NoError(t, reporter.ReportLog(false))
// Step 2: Close() updates state — sets Result=FAILURE and marks steps cancelled.
// In the real race, this happens while RunDaemon is between ReportLog and ReportState.
reporter.stateMu.Lock()
reporter.closed = true
for _, v := range reporter.state.Steps {
@@ -171,20 +147,13 @@ func TestReporter_EphemeralRunnerDeletion(t *testing.T) {
reporter.state.StoppedAt = timestamppb.Now()
reporter.stateMu.Unlock()
// Step 3: RunDaemon's ReportState() — with the fix, this returns early
// because closed=true, preventing the server from deleting the runner.
require.NoError(t, reporter.ReportState(false))
assert.False(t, runnerDeleted, "runner must not be deleted by RunDaemon's ReportState")
// Step 4: Close's final log upload succeeds because the runner is still alive.
// Flush pending rows first, then send the noMore signal (matching Close's retry behavior).
require.NoError(t, reporter.ReportLog(false))
// Acknowledge Close as done in daemon
close(reporter.daemon)
err = reporter.ReportLog(true)
require.NoError(t, err, "final log upload must not fail: runner should not be deleted before Close finishes sending logs")
err = reporter.ReportState(true)
require.NoError(t, err, "final state update should work: runner should not be deleted before Close finishes sending logs")
require.NoError(t, reporter.ReportLog(true), "final log upload must not fail: runner should not be deleted before Close finishes sending logs")
require.NoError(t, reporter.ReportState(true), "final state update should work: runner should not be deleted before Close finishes sending logs")
}
func TestReporter_RunDaemonClose_Race(t *testing.T) {
@@ -211,125 +180,56 @@ func TestReporter_RunDaemonClose_Race(t *testing.T) {
}, cfg)
reporter.ResetSteps(newTestSteps(1))
// Start the daemon loop — RunDaemon spawns a goroutine internally.
reporter.RunDaemon()
// Close concurrently — this races with the daemon goroutine on r.closed.
require.NoError(t, reporter.Close(""))
// Cancel context so the daemon goroutine exits cleanly.
cancel()
}
// TestReporter_MaxLatencyTimer verifies that the maxLatencyTimer flushes a
// single buffered log row before the periodic logTicker fires.
//
// Setup: logReportInterval=10s (effectively never), maxLatency=100ms.
// Fire one log line, then assert UpdateLog is called within 500ms.
func TestReporter_MaxLatencyTimer(t *testing.T) {
var updateLogCalls atomic.Int64
func TestReporter_LogFlush(t *testing.T) {
for _, tc := range []struct {
name string
maxLatency time.Duration
batchSize int
messages []string
}{
{"max latency timer", 100 * time.Millisecond, 1000, []string{"single log line"}},
{"batch size", 10 * time.Second, 5, []string{"log line 0", "log line 1", "log line 2", "log line 3", "log line 4"}},
} {
t.Run(tc.name, func(t *testing.T) {
var updateLogCalls atomic.Int64
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Return(
func(_ctx context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
updateLogCalls.Add(1)
return &runnerv1.UpdateLogResponse{AckIndex: req.Index + int64(len(req.Rows))}, nil
},
)
client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(&runnerv1.UpdateTaskResponse{}, nil)
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Return(
func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
updateLogCalls.Add(1)
return &runnerv1.UpdateLogResponse{
AckIndex: req.Index + int64(len(req.Rows)),
}, nil
},
)
client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(
func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
return &runnerv1.UpdateTaskResponse{}, nil
},
)
cfg, _ := config.LoadDefault("")
cfg.Runner.LogReportInterval = 10 * time.Second
cfg.Runner.LogReportMaxLatency = tc.maxLatency
cfg.Runner.LogReportBatchSize = tc.batchSize
reporter := newTestReporter(t, client, cfg)
reporter.ResetSteps(newTestSteps(1))
reporter.RunDaemon()
defer func() { require.NoError(t, reporter.Close("")) }()
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
// Custom config: logTicker=10s (won't fire during test), maxLatency=100ms
cfg, _ := config.LoadDefault("")
cfg.Runner.LogReportInterval = 10 * time.Second
cfg.Runner.LogReportMaxLatency = 100 * time.Millisecond
cfg.Runner.LogReportBatchSize = 1000 // won't trigger batch flush
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(newTestSteps(1))
reporter.RunDaemon()
defer func() {
_ = reporter.Close("")
}()
// Fire a single log line — not enough to trigger batch flush
require.NoError(t, fire(reporter, logEntry{
Message: "single log line",
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
// maxLatencyTimer should flush within ~100ms. Wait up to 500ms.
assert.Eventually(t, func() bool {
return updateLogCalls.Load() > 0
}, 500*time.Millisecond, 10*time.Millisecond,
"maxLatencyTimer should have flushed the log before logTicker (10s)")
}
// TestReporter_BatchSizeFlush verifies that reaching logBatchSize triggers
// an immediate log flush without waiting for any timer.
func TestReporter_BatchSizeFlush(t *testing.T) {
var updateLogCalls atomic.Int64
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Return(
func(_ context.Context, req *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
updateLogCalls.Add(1)
return &runnerv1.UpdateLogResponse{
AckIndex: req.Index + int64(len(req.Rows)),
}, nil
},
)
client.On("UpdateTask", mock.Anything, mock.Anything).Maybe().Return(
func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
return &runnerv1.UpdateTaskResponse{}, nil
},
)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
// Custom config: large timers, small batch size
cfg, _ := config.LoadDefault("")
cfg.Runner.LogReportInterval = 10 * time.Second
cfg.Runner.LogReportMaxLatency = 10 * time.Second
cfg.Runner.LogReportBatchSize = 5
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(newTestSteps(1))
reporter.RunDaemon()
defer func() {
_ = reporter.Close("")
}()
// Fire exactly batchSize log lines
for i := range 5 {
require.NoError(t, fire(reporter, logEntry{
Message: fmt.Sprintf("log line %d", i),
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
for _, message := range tc.messages {
require.NoError(t, fire(reporter, logEntry{
Message: message,
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
}
assert.Eventually(t, func() bool {
return updateLogCalls.Load() > 0
}, 500*time.Millisecond, 10*time.Millisecond, "logs must flush before the periodic timers")
})
}
// Batch threshold should trigger immediate flush
assert.Eventually(t, func() bool {
return updateLogCalls.Load() > 0
}, 500*time.Millisecond, 10*time.Millisecond,
"batch size threshold should have triggered immediate flush")
}
// TestReporter_StateChangedNotLostDuringReport asserts that a log record arriving
// mid-UpdateTask re-dirties the flag so the change is picked up by the next report.
func TestReporter_StateChangedNotLostDuringReport(t *testing.T) {
var updateTaskCalls atomic.Int64
inFlight := make(chan struct{})
@@ -340,7 +240,6 @@ func TestReporter_StateChangedNotLostDuringReport(t *testing.T) {
func(_ context.Context, _ *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
n := updateTaskCalls.Add(1)
if n == 1 {
// Signal that the first UpdateTask is in flight, then block until released.
close(inFlight)
<-release
}
@@ -348,53 +247,38 @@ func TestReporter_StateChangedNotLostDuringReport(t *testing.T) {
},
)
ctx, cancel := context.WithCancel(t.Context())
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 := newTestReporter(t, client, cfg)
reporter.ResetSteps(newTestSteps(2))
// Mark stateChanged=true so the first ReportState proceeds to UpdateTask.
reporter.stateMu.Lock()
reporter.stateChanged = true
reporter.stateMu.Unlock()
// Kick off the first ReportState in a goroutine — it will block in UpdateTask.
done := make(chan error, 1)
go func() {
done <- reporter.ReportState(false)
}()
// Wait until UpdateTask is in flight (snapshot taken, flag consumed).
<-inFlight
// A concurrent log record modifies state — must re-flip stateChanged so the
// change is not lost when the in-flight ReportState finishes.
require.NoError(t, fire(reporter, logEntry{
Message: "step starts",
Data: map[string]any{"stage": "Main", "stepNumber": 1, "raw_output": true},
}))
// Release the in-flight UpdateTask and wait for it to return.
close(release)
require.NoError(t, <-done)
// stateChanged must still be true so the next ReportState picks up the
// concurrent record's change instead of skipping via the early-return path.
reporter.stateMu.RLock()
changed := reporter.stateChanged
reporter.stateMu.RUnlock()
assert.True(t, changed, "stateChanged must remain true after a concurrent log record during in-flight ReportState")
// And the next ReportState must actually send a second UpdateTask.
require.NoError(t, reporter.ReportState(false))
assert.Equal(t, int64(2), updateTaskCalls.Load(), "concurrent log record change must trigger a second UpdateTask, not be silently lost")
}
// TestReporter_StateChangedRestoredOnError verifies that when UpdateTask fails,
// the dirty flag is restored so the snapshotted change isn't silently lost.
func TestReporter_StateChangedRestoredOnError(t *testing.T) {
var updateTaskCalls atomic.Int64
@@ -409,19 +293,14 @@ func TestReporter_StateChangedRestoredOnError(t *testing.T) {
},
)
ctx, cancel := context.WithCancel(t.Context())
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 := newTestReporter(t, client, cfg)
reporter.ResetSteps(newTestSteps(1))
reporter.stateMu.Lock()
reporter.stateChanged = true
reporter.stateMu.Unlock()
// First ReportState fails — flag must be restored to true.
require.Error(t, reporter.ReportState(false))
reporter.stateMu.RLock()
@@ -429,13 +308,10 @@ func TestReporter_StateChangedRestoredOnError(t *testing.T) {
reporter.stateMu.RUnlock()
assert.True(t, changed, "stateChanged must be restored to true after UpdateTask error so the change is retried")
// The next ReportState should still issue a request because the flag was restored.
require.NoError(t, reporter.ReportState(false))
assert.Equal(t, int64(2), updateTaskCalls.Load())
}
// TestReporter_StateNotifyFlush verifies that step transitions trigger
// an immediate state flush via the stateNotify channel.
func TestReporter_StateNotifyFlush(t *testing.T) {
var updateTaskCalls atomic.Int64
@@ -454,41 +330,28 @@ func TestReporter_StateNotifyFlush(t *testing.T) {
},
)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
// Custom config: large state interval so only stateNotify can trigger
cfg, _ := config.LoadDefault("")
cfg.Runner.StateReportInterval = 10 * time.Second
cfg.Runner.LogReportInterval = 10 * time.Second
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter := newTestReporter(t, client, cfg)
reporter.ResetSteps(newTestSteps(1))
reporter.RunDaemon()
defer func() {
_ = reporter.Close("")
}()
// Fire a log entry that starts a step — this triggers stateNotify
require.NoError(t, fire(reporter, logEntry{
Message: "step starting",
Data: map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true},
}))
// stateNotify should trigger immediate UpdateTask call
assert.Eventually(t, func() bool {
return updateTaskCalls.Load() > 0
}, 500*time.Millisecond, 10*time.Millisecond,
"step transition should have triggered immediate state flush via stateNotify")
}
// Regression test for https://gitea.com/gitea/runner/issues/950: Close() must
// always send a final UpdateLog with NoMore=true carrying at least one row,
// otherwise the server's len(Rows)==0 short-circuit skips TransferLogs.
// TODO: Remove after https://github.com/go-gitea/gitea/pull/37631 is in all
// supported branches, e.g. v1.28+.
func TestReporter_CloseAlwaysSendsRowsWithNoMore(t *testing.T) {
var lastReq atomic.Pointer[runnerv1.UpdateLogRequest]
var noMoreCalls atomic.Int64
@@ -511,25 +374,16 @@ func TestReporter_CloseAlwaysSendsRowsWithNoMore(t *testing.T) {
},
)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
// Intervals large enough that no daemon-driven flush fires during the test.
cfg, _ := config.LoadDefault("")
cfg.Runner.LogReportInterval = 10 * time.Second
cfg.Runner.LogReportMaxLatency = 10 * time.Second
cfg.Runner.StateReportInterval = 10 * time.Second
cfg.Runner.LogReportBatchSize = 1000
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter := newTestReporter(t, client, cfg)
reporter.ResetSteps(newTestSteps(1))
reporter.RunDaemon()
// Simulate a successful job whose log buffer was already drained by the
// daemon (logOffset > 0, logRows empty, terminal Result set). This is the
// state Close() lands in for the typical successful job under #819.
reporter.stateMu.Lock()
reporter.logOffset = 5
reporter.state.Result = runnerv1.Result_RESULT_SUCCESS
@@ -546,11 +400,6 @@ func TestReporter_CloseAlwaysSendsRowsWithNoMore(t *testing.T) {
assert.NotEmpty(t, final.Rows, "final UpdateLog must carry at least one row")
}
// TestReporter_StateHeartbeat verifies that ReportState sends a heartbeat
// UpdateTask once stateReportInterval has elapsed since the last successful
// report, even when nothing has changed. Without this, long-running silent
// jobs (no log output, no step transitions) cause the server to time the
// task out and cancel it (#826).
func TestReporter_StateHeartbeat(t *testing.T) {
var updateTaskCalls atomic.Int64
@@ -562,39 +411,25 @@ func TestReporter_StateHeartbeat(t *testing.T) {
},
)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
cfg, _ := config.LoadDefault("")
cfg.Runner.StateReportInterval = 50 * time.Millisecond
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter := newTestReporter(t, client, cfg)
reporter.ResetSteps(newTestSteps(1))
// First call has no prior report — sends to seed lastReportedAt.
reporter.stateMu.Lock()
reporter.stateChanged = true
reporter.stateMu.Unlock()
require.NoError(t, reporter.ReportState(false))
require.Equal(t, int64(1), updateTaskCalls.Load())
// Second call immediately after with nothing changed — must skip.
require.NoError(t, reporter.ReportState(false))
assert.Equal(t, int64(1), updateTaskCalls.Load(), "no-op ReportState within stateReportInterval must skip")
// After stateReportInterval elapses, a heartbeat must fire even with no changes.
time.Sleep(2 * cfg.Runner.StateReportInterval)
require.NoError(t, reporter.ReportState(false))
assert.Equal(t, int64(2), updateTaskCalls.Load(), "ReportState must heartbeat after stateReportInterval even with no state change")
}
// TestReporter_ServerCancelStillFlushesFinal asserts that when the Gitea server
// returns RESULT_CANCELLED on an in-flight UpdateTask (which causes the
// reporter to cancel the task context), Close() still successfully sends the
// final UpdateLog{NoMore:true} and the final UpdateTask carrying the populated
// final state. Before the fix this final flush used r.ctx, which was just
// cancelled, so retry-go aborted on its context check and Gitea never received
// the runner's acknowledgement of the cancel.
func TestReporter_ServerCancelStillFlushesFinal(t *testing.T) {
var (
updateTaskCalls atomic.Int64
@@ -613,10 +448,6 @@ func TestReporter_ServerCancelStillFlushesFinal(t *testing.T) {
}, nil
},
)
// The first UpdateTask returns RESULT_CANCELLED — modelling a server-side
// cancellation; the reporter must call r.cancel() in response. The final
// UpdateTask issued by Close() must still arrive even though r.ctx is now
// cancelled.
client.On("UpdateTask", mock.Anything, mock.Anything).Return(
func(_ context.Context, req *runnerv1.UpdateTaskRequest) (*runnerv1.UpdateTaskResponse, error) {
n := updateTaskCalls.Add(1)
@@ -642,42 +473,27 @@ func TestReporter_ServerCancelStillFlushesFinal(t *testing.T) {
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(newTestSteps(1))
// Force the first ReportState to actually call UpdateTask.
reporter.stateMu.Lock()
reporter.stateChanged = true
reporter.stateMu.Unlock()
// First ReportState — server returns RESULT_CANCELLED, reporter cancels r.ctx.
require.NoError(t, reporter.ReportState(false))
require.Equal(t, int64(1), updateTaskCalls.Load())
select {
case <-ctx.Done():
// Expected: reporter called cancel() because the server reported the task as cancelled.
case <-time.After(time.Second):
t.Fatal("expected r.ctx to be cancelled after server returned RESULT_CANCELLED")
}
// The test does not start the daemon goroutine; close(r.daemon) so Close()
// proceeds without waiting on its 60s timeout.
close(reporter.daemon)
// Now Close() runs. Before the fix, both final RPCs aborted on the cancelled
// r.ctx via retry.Context. After the fix, Close() uses a detached context and
// the per-RPC rpcCtx() falls back to a fresh ctx, so both calls succeed.
require.NoError(t, reporter.Close("cancelled"))
assert.True(t, finalLogNoMoreSeen.Load(), "Close() must send a final UpdateLog{NoMore:true} even after server-side cancellation")
assert.True(t, finalTaskStateSeen.Load(), "Close() must send a final UpdateTask with the populated final state even after server-side cancellation")
}
// TestReporter_CloseReportsCancelledOnCanceledCtx asserts that when Close()
// runs on a reporter whose state has not been finalised AND whose context has
// been cancelled, the synthesised final state carries RESULT_CANCELLED and
// the appended log row reads "Cancelled" — not RESULT_FAILURE / "Early
// termination". This is the runner-side half of the Running -> Cancelling ->
// Cancelled flow: it gives Gitea an explicit cancel acknowledgement rather
// than a generic failure when the job is torn down on the cancel path.
func TestReporter_CloseReportsCancelledOnCanceledCtx(t *testing.T) {
var finalState atomic.Pointer[runnerv1.TaskState]
var finalLogRows atomic.Pointer[[]*runnerv1.LogRow]
@@ -710,13 +526,10 @@ func TestReporter_CloseReportsCancelledOnCanceledCtx(t *testing.T) {
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter.ResetSteps(newTestSteps(1))
// Simulate the cancellation path: r.ctx is cancelled before Close() runs.
cancel()
// Skip the daemon wait inside Close().
close(reporter.daemon)
// Empty lastWords so Close() picks the synthesised value.
require.NoError(t, reporter.Close(""))
got := finalState.Load()
@@ -742,8 +555,6 @@ func TestReporter_CloseReportsCancelledOnCanceledCtx(t *testing.T) {
assert.False(t, foundEarlyTermination, "final log must not contain 'Early termination' on the cancel path")
}
// TestReporter_StopHeartbeats verifies that StopHeartbeats ends periodic
// UpdateTask heartbeats while Close() still flushes the final state.
func TestReporter_StopHeartbeats(t *testing.T) {
var updateTaskCalls atomic.Int64
@@ -762,17 +573,12 @@ func TestReporter_StopHeartbeats(t *testing.T) {
},
)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
cfg, err := config.LoadDefault("")
require.NoError(t, err)
cfg.Runner.StateReportInterval = 20 * time.Millisecond
cfg.Runner.LogReportInterval = time.Hour
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter := newTestReporter(t, client, cfg)
reporter.ResetSteps(newTestSteps(1))
reporter.RunDaemon()
@@ -810,7 +616,6 @@ func TestAppendLogRow(t *testing.T) {
require.Len(t, r.logRows, 1)
assert.Equal(t, "hello ***", r.logRows[0].Content)
// repairing the invalid byte spells out the secret, so the repair has to come first
r = &Reporter{logReplacer: strings.NewReplacer("a?b", "***")}
r.appendLogRow(&runnerv1.LogRow{Time: timestamppb.Now(), Content: "a\xffb"})
assert.Equal(t, "***", r.logRows[0].Content)
@@ -835,34 +640,28 @@ func TestReporter_SetOutputs(t *testing.T) {
require.True(t, ok)
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"})
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[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["big"]
assert.False(t, ok)
// a value carrying a secret is skipped, as GitHub does, rather than sent masked
r.SetOutputs(map[string]string{"leaky": "has s3cr3t in it"})
_, ok = r.outputs["leaky"]
assert.False(t, ok)
// invalid UTF-8 is not a secret, so the value is kept as it is rather than dropped
r.SetOutputs(map[string]string{"binary": "caf\xff"})
got, ok = r.outputs["binary"]
require.True(t, ok)
assert.Equal(t, "caf\xff", got.value)
// 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["atlimit"]
@@ -870,8 +669,6 @@ func TestReporter_SetOutputs(t *testing.T) {
assert.Len(t, got.value, maxOutputValueLen)
}
// Gitea delivers ACTIONS_STEP_DEBUG as a secret, so masking "true" would drop any job output
// saying it. GitHub skips the same two keys.
func TestReporter_DebugSettingsAreNotMasked(t *testing.T) {
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
@@ -886,10 +683,9 @@ func TestReporter_DebugSettingsAreNotMasked(t *testing.T) {
assert.Equal(t, "***", reporter.mask("s3cr3t"))
reporter.SetOutputs(map[string]string{"changed": "true"})
assert.Equal(t, "true", reporter.outputs["changed"].value) // needs.<job>.outputs.changed == 'true' still works
assert.Equal(t, "true", reporter.outputs["changed"].value)
}
// An output the server acknowledged is not reported again.
func TestReporter_OutputsSentOnce(t *testing.T) {
client := mocks.NewClient(t)
var reported []map[string]string
@@ -899,12 +695,8 @@ func TestReporter_OutputsSentOnce(t *testing.T) {
return &runnerv1.UpdateTaskResponse{SentOutputs: slices.Collect(maps.Keys(req.Outputs))}, nil
})
ctx, cancel := context.WithCancel(t.Context())
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 := newTestReporter(t, client, cfg)
r.SetOutputs(map[string]string{"foo": "bar"})
require.NoError(t, r.ReportState(false))
@@ -935,8 +727,6 @@ func TestReporter_FlushFinalRetriesUntilSuccess(t *testing.T) {
}
func TestReporter_ParseResult(t *testing.T) {
r := &Reporter{}
tests := []struct {
name string
input any
@@ -951,15 +741,13 @@ func TestReporter_ParseResult(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, ok := r.parseResult(tt.input)
got, ok := parseResult(tt.input)
assert.Equal(t, tt.wantOk, ok)
assert.Equal(t, tt.want, got)
})
}
}
// A secret leaked in an encoded form — the shape it takes once an action puts it in a
// JSON body, a URL or a base64 payload — must be masked in the reported log as well.
func TestReporter_masksEncodedSecrets(t *testing.T) {
secret := `p@ss w"rd/1`
r := &Reporter{logReplacer: strings.NewReplacer()}
@@ -978,20 +766,14 @@ func TestReporter_masksEncodedSecrets(t *testing.T) {
}
}
// 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, _ *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
return &runnerv1.UpdateLogResponse{AckIndex: 1000}, nil
})
ctx, cancel := context.WithCancel(t.Context())
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 := newTestReporter(t, client, cfg)
reporter.ResetSteps(newTestSteps(1))
require.NoError(t, fire(reporter, logEntry{
@@ -1003,7 +785,6 @@ func TestReporter_AckIndexBeyondBuffer(t *testing.T) {
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{})
@@ -1023,13 +804,9 @@ func TestReporter_CloseWithStuckDaemon(t *testing.T) {
return &runnerv1.UpdateTaskResponse{}, nil
}).Maybe()
ctx, cancel := context.WithCancel(t.Context())
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 := newTestReporter(t, client, cfg)
reporter.daemonWait = time.Millisecond
reporter.RunDaemon()
reporter.ResetSteps(newTestSteps(1))
@@ -1053,8 +830,6 @@ func TestReporter_CloseWithStuckDaemon(t *testing.T) {
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)
@@ -1076,7 +851,7 @@ func TestReporter_ServerFailureResultIsReportedAsFailure(t *testing.T) {
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
close(reporter.daemon)
require.NoError(t, reporter.ReportState(false))
require.ErrorIs(t, ctx.Err(), context.Canceled)
@@ -1091,11 +866,9 @@ func TestReporter_NoteReport(t *testing.T) {
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.AllRecords())
// 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.AllRecords(), 1)
@@ -1106,8 +879,6 @@ func TestReporter_NoteReport(t *testing.T) {
assert.Contains(t, hook.LastRecord().Message, "reconnected")
}
// A job's final error can carry a secret, e.g. one interpolated into a failing
// expression, so Close must mask it like every other row.
func TestReporter_CloseMasksLastWords(t *testing.T) {
const secret = "supersecret"
var rows []*runnerv1.LogRow
@@ -1145,9 +916,6 @@ func TestReporter_CloseMasksLastWords(t *testing.T) {
assert.Equal(t, "could not get job matrix: *** ***", rows[0].Content)
}
// giteaLogModel mirrors how Gitea stores a task log: UpdateLog appends rows to one
// stream, and UpdateTask overwrites the per-step ranges the web UI slices that stream by
// (modules/actions/task_state.go, FullSteps).
type giteaLogModel struct {
mu sync.Mutex
rows int64
@@ -1169,15 +937,12 @@ func (m *giteaLogModel) applyState(state *runnerv1.TaskState) {
}
}
// snapshot returns the accepted rows and how far the reported step ranges reach into
// them. The remainder is what the web UI renders under "Complete job".
func (m *giteaLogModel) snapshot() (rows, claimed int64) {
m.mu.Lock()
defer m.mu.Unlock()
return m.rows, m.claimed
}
// Regression test for https://gitea.com/gitea/runner/issues/1184.
func TestReporter_StepRangeCoversAckedRows(t *testing.T) {
server := &giteaLogModel{}
@@ -1196,20 +961,13 @@ func TestReporter_StepRangeCoversAckedRows(t *testing.T) {
},
)
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
// Only the batch threshold may flush: the tickers are the periodic repair this test
// must not depend on, and Close() is the one after the damage is done.
cfg, _ := config.LoadDefault("")
cfg.Runner.LogReportInterval = time.Hour
cfg.Runner.LogReportMaxLatency = time.Hour
cfg.Runner.StateReportInterval = time.Hour
cfg.Runner.LogReportBatchSize = 5
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter := newTestReporter(t, client, cfg)
reporter.ResetSteps(newTestSteps(1))
reporter.RunDaemon()
defer func() {
@@ -1218,7 +976,6 @@ func TestReporter_StepRangeCoversAckedRows(t *testing.T) {
stepData := map[string]any{"stage": "Main", "stepNumber": 0, "raw_output": true}
// The step's first row is a transition, so it flushes state as well as logs.
require.NoError(t, fire(reporter, logEntry{Message: "step starting", Data: stepData}))
require.Eventually(t, func() bool {
rows, claimed := server.snapshot()
@@ -1239,14 +996,7 @@ func TestReporter_StepRangeCoversAckedRows(t *testing.T) {
"every log row the server accepted must be covered by a reported step range")
}
// A flush the server took no rows for describes nothing new, so it must not spend an
// UpdateTask. UpdateTask is never registered on the client, so calling it fails the test.
func TestReporter_FlushWithoutAckedRowsSkipsState(t *testing.T) {
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
client := mocks.NewClient(t)
client.On("UpdateLog", mock.Anything, mock.Anything).Once().Return(
func(_ context.Context, _ *runnerv1.UpdateLogRequest) (*runnerv1.UpdateLogResponse, error) {
@@ -1255,16 +1005,23 @@ func TestReporter_FlushWithoutAckedRowsSkipsState(t *testing.T) {
)
cfg, _ := config.LoadDefault("")
reporter := NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
reporter := newTestReporter(t, client, cfg)
// An idle job: no rows, so not even an UpdateLog.
reporter.reportLogWithState()
// A row the server declines to take: an UpdateLog, but still no step range to report.
reporter.Logf("row")
reporter.reportLogWithState()
}
func newTestReporter(t *testing.T, client *mocks.Client, cfg *config.Config) *Reporter {
t.Helper()
ctx, cancel := context.WithCancel(t.Context())
t.Cleanup(cancel)
taskCtx, err := structpb.NewStruct(map[string]any{})
require.NoError(t, err)
return NewReporter(ctx, cancel, client, &runnerv1.Task{Context: taskCtx}, cfg)
}
func newTestSteps(n int) []*model.Step {
steps := make([]*model.Step, n)
for i := range steps {
@@ -1353,7 +1110,7 @@ func TestReporter_ReportedSteps(t *testing.T) {
{main, 0, "Echo 1", success, 6, 2},
{main, 1, "Run echo 1", skipped, 0, 0},
{post, 0, "Post Echo 1", success, 8, 3},
{cleanup, 0, "Complete job", unspecified, 11, 2}, // still logging after the job result
{cleanup, 0, "Complete job", unspecified, 11, 2},
}
assert.Equal(t, want, reportedOf(t, reporter))
@@ -1362,7 +1119,6 @@ func TestReporter_ReportedSteps(t *testing.T) {
want[5].result, want[5].logLen = success, 3
assert.Equal(t, want, reportedOf(t, reporter))
// the workflow steps keep what Gitea versions deriving them from the workflow file expect
require.Len(t, reporter.state.Steps, 2)
assert.EqualValues(t, 6, reporter.state.Steps[0].LogIndex)
assert.EqualValues(t, 2, reporter.state.Steps[0].LogLength)
@@ -1403,7 +1159,7 @@ func TestReporter_ReportedSteps(t *testing.T) {
"Set up job": success,
"Pre Run echo 0": success,
"Pre Run echo 2": success,
"Run echo 0": failure, // not the result of its composite step
"Run echo 0": failure,
"Run echo 1": cancelled,
"Run echo 2": cancelled,
"Post Run echo 0": failure,
@@ -1473,7 +1229,6 @@ func TestReporter_ReportedSteps(t *testing.T) {
})
}
// logEntry is the record a job logger hands the Reporter, with the fields it accumulated as Data.
type logEntry struct {
Message string
Data map[string]any