Files
act_runner/pkg/common/executor.go
Pascal Zimmermann c0f19d9a26 fix: Max parallel Support for Matrix Jobs and Remote Action Tests (#150)
## Summary

This PR fixes the `max-parallel` strategy configuration for matrix jobs and resolves all failing tests in `step_action_remote_test.go`. The implementation ensures that matrix jobs respect the `max-parallel` setting, preventing resource exhaustion when running GitHub Actions workflows.

## Problem Statement

### Issue 1: max-parallel Not Working Correctly
Matrix jobs were running in parallel regardless of the `max-parallel` setting in the strategy configuration. This caused:
- Resource contention on limited runners
- Unpredictable job execution behavior
- Inability to control concurrency for resource-intensive workflows

### Issue 2: Failing Remote Action Tests
All tests in `step_action_remote_test.go` were failing due to:
- Missing `ActionCacheDir` configuration
- Incorrect mock expectations using fixed strings instead of hash-based paths
- Incompatibility with the hash-based action cache implementation

## Changes

### 1. max-parallel Implementation (`pkg/runner/runner.go`)

#### Robust Initialization
Added fallback logic to ensure `MaxParallel` is always properly initialized:
```go
if job.Strategy.MaxParallel == 0 {
    job.Strategy.MaxParallel = job.Strategy.GetMaxParallel()
}
```

#### Eliminated Unnecessary Nesting
Fixed inefficient nested parallelization when only one pipeline element exists:
```go
if len(pipeline) == 1 {
    // Execute directly without additional wrapper
    log.Debugf("Single pipeline element, executing directly")
    return pipeline[0](ctx)
}
```

#### Enhanced Logging
Added comprehensive debug and info logging:
- Shows which `maxParallel` value is being used
- Logs adjustments based on matrix size
- Reports final parallelization decisions

### 2. Worker Logging (`pkg/common/executor.go`)

Enhanced `NewParallelExecutor` with detailed worker activity logging:
```go
log.Infof("NewParallelExecutor: Creating %d workers for %d executors", parallel, len(executors))

for i := 0; i < parallel; i++ {
    go func(workerID int, work <-chan Executor, errs chan<- error) {
        log.Debugf("Worker %d started", workerID)
        taskCount := 0
        for executor := range work {
            taskCount++
            log.Debugf("Worker %d executing task %d", workerID, taskCount)
            errs <- executor(ctx)
        }
        log.Debugf("Worker %d finished (%d tasks executed)", workerID, taskCount)
    }(i, work, errs)
}
```

**Benefits:**
- Easy verification of correct parallelization
- Better debugging capabilities
- Clear visibility into worker activity

### 3. Documentation (`pkg/model/workflow.go`)

Added clarifying comment to ensure strategy values are always set:
```go
// Always set these values, even if there's an error later
j.Strategy.FailFast = j.Strategy.GetFailFast()
j.Strategy.MaxParallel = j.Strategy.GetMaxParallel()
```

### 4. Test Fixes (`pkg/runner/step_action_remote_test.go`)

#### Added Missing Configuration
All tests now include `ActionCacheDir`:
```go
Config: &Config{
    GitHubInstance: "github.com",
    ActionCacheDir: "/tmp/test-cache",
}
```

#### Hash-Based Suffix Matchers
Updated mocks to use hash-based paths instead of fixed strings:
```go
// Before
sarm.On("readAction", sar.Step, suffixMatcher("org-repo-path@ref"), ...)

// After
sarm.On("readAction", sar.Step, suffixMatcher(sar.Step.UsesHash()), ...)
```

#### Flexible Exec Matcher for Post Tests
Implemented flexible path matching for hash-based action directories:
```go
execMatcher := mock.MatchedBy(func(args []string) bool {
    if len(args) != 2 {
        return false
    }
    return args[0] == "node" && strings.Contains(args[1], "post.js")
})
```

#### Token Test Improvements
- Uses unique cache directory to force cloning
- Validates URL redirection to github.com
- Accepts realistic token behavior

### 5. New Tests

#### Unit Tests (`pkg/runner/max_parallel_test.go`)
Tests various `max-parallel` configurations:
- `max-parallel: 1` → Sequential execution
- `max-parallel: 2` → Max 2 parallel jobs
- `max-parallel: 4` (default) → Max 4 parallel jobs
- `max-parallel: 10` → Max 10 parallel jobs

#### Concurrency Test (`pkg/common/executor_max_parallel_test.go`)
Verifies that `max-parallel: 2` actually limits concurrent execution using atomic counters.

## Expected Behavior

### Before
-  Jobs ran in parallel regardless of `max-parallel` setting
-  Unnecessary nested parallelization (8 workers for 1 element)
-  No logging to debug parallelization issues
-  All `step_action_remote_test.go` tests failing

### After
-  `max-parallel: 1` → Jobs run strictly sequentially
-  `max-parallel: N` → Maximum N jobs run simultaneously
-  Efficient single-level parallelization for matrix jobs
-  Comprehensive logging for debugging
-  All tests passing (10/10)

## Example Log Output

With `max-parallel: 2` and 6 matrix jobs:
```
[DEBUG] Using job.Strategy.MaxParallel: 2
[INFO] Running job with maxParallel=2 for 6 matrix combinations
[DEBUG] Single pipeline element, executing directly
[INFO] NewParallelExecutor: Creating 2 workers for 6 executors
[DEBUG] Worker 0 started
[DEBUG] Worker 1 started
[DEBUG] Worker 0 executing task 1
[DEBUG] Worker 1 executing task 1
...
[DEBUG] Worker 0 finished (3 tasks executed)
[DEBUG] Worker 1 finished (3 tasks executed)
```

## Test Results

All tests pass successfully:
```
 TestStepActionRemote (3 sub-tests)
 TestStepActionRemotePre
 TestStepActionRemotePreThroughAction
 TestStepActionRemotePreThroughActionToken
 TestStepActionRemotePost (4 sub-tests)
 TestMaxParallelStrategy
 TestMaxParallel2Quick

Total: 12/12 tests passing
```

## Breaking Changes

None. This PR is fully backward compatible. All changes improve existing behavior without altering the API.

## Impact

-  Fixes resource management for CI/CD pipelines
-  Prevents runner exhaustion on limited infrastructure
-  Enables sequential execution for resource-intensive jobs
-  Improves debugging with detailed logging
-  Ensures test suite reliability

## Files Modified

### Core Implementation
- `pkg/runner/runner.go` - max-parallel fix + logging
- `pkg/common/executor.go` - Worker logging
- `pkg/model/workflow.go` - Documentation

### Tests
- `pkg/runner/step_action_remote_test.go` - Fixed all 10 tests
- `pkg/runner/max_parallel_test.go` - **NEW** - Unit tests
- `pkg/common/executor_max_parallel_test.go` - **NEW** - Concurrency test

## Verification

### Manual Testing
```bash
# Build
go build -o dist/act main.go

# Test with max-parallel: 2
./dist/act -W test-max-parallel-2.yml -v

# Expected: Only 2 jobs run simultaneously
```

### Automated Testing
```bash
# Run all tests
go test ./pkg/runner -run TestStepActionRemote -v
go test ./pkg/runner -run TestMaxParallel -v
go test ./pkg/common -run TestMaxParallel -v
```

## Related Issues

Fixes issues where matrix jobs in Gitea ignored the `max-parallel` strategy setting, causing resource contention and unpredictable behavior.

Reviewed-on: https://gitea.com/gitea/act/pulls/150
Reviewed-by: Lunny Xiao <xiaolunwen@gmail.com>
Reviewed-by: silverwind <silverwind@noreply.gitea.com>
Co-authored-by: Pascal Zimmermann <pascal.zimmermann@theiotstudio.com>
Co-committed-by: Pascal Zimmermann <pascal.zimmermann@theiotstudio.com>
2026-02-11 00:43:38 +00:00

204 lines
4.7 KiB
Go

package common
import (
"context"
"fmt"
log "github.com/sirupsen/logrus"
)
// Warning that implements `error` but safe to ignore
type Warning struct {
Message string
}
// Error the contract for error
func (w Warning) Error() string {
return w.Message
}
// Warningf create a warning
func Warningf(format string, args ...interface{}) Warning {
w := Warning{
Message: fmt.Sprintf(format, args...),
}
return w
}
// Executor define contract for the steps of a workflow
type Executor func(ctx context.Context) error
// Conditional define contract for the conditional predicate
type Conditional func(ctx context.Context) bool
// NewInfoExecutor is an executor that logs messages
func NewInfoExecutor(format string, args ...interface{}) Executor {
return func(ctx context.Context) error {
logger := Logger(ctx)
logger.Infof(format, args...)
return nil
}
}
// NewDebugExecutor is an executor that logs messages
func NewDebugExecutor(format string, args ...interface{}) Executor {
return func(ctx context.Context) error {
logger := Logger(ctx)
logger.Debugf(format, args...)
return nil
}
}
// NewPipelineExecutor creates a new executor from a series of other executors
func NewPipelineExecutor(executors ...Executor) Executor {
if len(executors) == 0 {
return func(ctx context.Context) error {
return nil
}
}
var rtn Executor
for _, executor := range executors {
if rtn == nil {
rtn = executor
} else {
rtn = rtn.Then(executor)
}
}
return rtn
}
// NewConditionalExecutor creates a new executor based on conditions
func NewConditionalExecutor(conditional Conditional, trueExecutor Executor, falseExecutor Executor) Executor {
return func(ctx context.Context) error {
if conditional(ctx) {
if trueExecutor != nil {
return trueExecutor(ctx)
}
} else {
if falseExecutor != nil {
return falseExecutor(ctx)
}
}
return nil
}
}
// NewErrorExecutor creates a new executor that always errors out
func NewErrorExecutor(err error) Executor {
return func(ctx context.Context) error {
return err
}
}
// NewParallelExecutor creates a new executor from a parallel of other executors
func NewParallelExecutor(parallel int, executors ...Executor) Executor {
return func(ctx context.Context) error {
work := make(chan Executor, len(executors))
errs := make(chan error, len(executors))
if 1 > parallel {
log.Debugf("Parallel tasks (%d) below minimum, setting to 1", parallel)
parallel = 1
}
log.Infof("NewParallelExecutor: Creating %d workers for %d executors", parallel, len(executors))
for i := 0; i < parallel; i++ {
go func(workerID int, work <-chan Executor, errs chan<- error) {
log.Debugf("Worker %d started", workerID)
taskCount := 0
for executor := range work {
taskCount++
log.Debugf("Worker %d executing task %d", workerID, taskCount)
errs <- executor(ctx)
}
log.Debugf("Worker %d finished (%d tasks executed)", workerID, taskCount)
}(i, work, errs)
}
for i := 0; i < len(executors); i++ {
work <- executors[i]
}
close(work)
// Executor waits all executors to cleanup these resources.
var firstErr error
for i := 0; i < len(executors); i++ {
err := <-errs
if firstErr == nil {
firstErr = err
}
}
if err := ctx.Err(); err != nil {
return err
}
return firstErr
}
}
// Then runs another executor if this executor succeeds
func (e Executor) Then(then Executor) Executor {
return func(ctx context.Context) error {
err := e(ctx)
if err != nil {
switch err.(type) {
case Warning:
Logger(ctx).Warning(err.Error())
default:
return err
}
}
if ctx.Err() != nil {
return ctx.Err()
}
return then(ctx)
}
}
// If only runs this executor if conditional is true
func (e Executor) If(conditional Conditional) Executor {
return func(ctx context.Context) error {
if conditional(ctx) {
return e(ctx)
}
return nil
}
}
// IfNot only runs this executor if conditional is true
func (e Executor) IfNot(conditional Conditional) Executor {
return func(ctx context.Context) error {
if !conditional(ctx) {
return e(ctx)
}
return nil
}
}
// IfBool only runs this executor if conditional is true
func (e Executor) IfBool(conditional bool) Executor {
return e.If(func(ctx context.Context) bool {
return conditional
})
}
// Finally adds an executor to run after other executor
func (e Executor) Finally(finally Executor) Executor {
return func(ctx context.Context) error {
err := e(ctx)
err2 := finally(ctx)
if err2 != nil {
return fmt.Errorf("Error occurred running finally: %v (original error: %v)", err2, err)
}
return err
}
}
// Not return an inverted conditional
func (c Conditional) Not() Conditional {
return func(ctx context.Context) bool {
return !c(ctx)
}
}