diff --git a/act/artifactcache/handler.go b/act/artifactcache/handler.go
index 3d6df28e..46e17e9b 100644
--- a/act/artifactcache/handler.go
+++ b/act/artifactcache/handler.go
@@ -34,6 +34,7 @@ import (
"github.com/timshannon/bolthold"
"go.etcd.io/bbolt"
+ berrors "go.etcd.io/bbolt/errors"
)
const (
@@ -44,6 +45,8 @@ const (
// Short enough that a leaked URL is near-worthless; long enough to let the
// @actions/cache client download a big blob that was returned from /cache.
artifactURLTTL = 10 * time.Minute
+
+ dbOpenTimeout = 5 * time.Second
)
type credKey struct{}
@@ -85,6 +88,8 @@ type Handler struct {
server *http.Server
logger *logging.Logger
+ dbLock chan struct{} // queues store openers without polling bolt's file lock
+
gcing atomic.Bool
gcAt time.Time
@@ -142,6 +147,7 @@ func StartHandler(opts Options) (*Handler, error) {
internalSecret: opts.InternalSecret,
policy: opts.Policy.withDefaults(),
freeDisk: disk.FreeBytes,
+ dbLock: make(chan struct{}, 1),
}
if logger == nil {
@@ -418,16 +424,42 @@ func (h *Handler) Close() error {
return retErr
}
-func (h *Handler) openDB() (*bolthold.Store, error) {
- return bolthold.Open(filepath.Join(h.dir, "bolt.db"), 0o644, &bolthold.Options{
+type cacheDB struct {
+ *bolthold.Store
+ close func() error
+}
+
+func (db *cacheDB) Close() error {
+ return db.close()
+}
+
+func (h *Handler) openDB() (store *cacheDB, err error) {
+ select {
+ case h.dbLock <- struct{}{}:
+ case <-time.After(dbOpenTimeout):
+ return nil, berrors.ErrTimeout
+ }
+ defer func() {
+ if store == nil { // only an opened store hands the lock to Close, even when Open panics
+ <-h.dbLock
+ }
+ }()
+ db, err := bolthold.Open(filepath.Join(h.dir, "bolt.db"), 0o644, &bolthold.Options{
Encoder: func(value any) ([]byte, error) { return json.Marshal(value) },
Decoder: func(data []byte, value any) error { return json.Unmarshal(data, value) },
Options: &bbolt.Options{
- Timeout: 5 * time.Second,
+ Timeout: dbOpenTimeout,
NoGrowSync: bbolt.DefaultOptions.NoGrowSync,
FreelistType: bbolt.DefaultOptions.FreelistType,
},
})
+ if err != nil {
+ return nil, err
+ }
+ return &cacheDB{Store: db, close: sync.OnceValue(func() error {
+ defer func() { <-h.dbLock }()
+ return db.Close()
+ })}, nil
}
// syncMetadata mirrors the bucket into the store until ctx ends, so no request waits on S3 for it.
@@ -528,7 +560,7 @@ func (h *Handler) find(w http.ResponseWriter, r *http.Request) {
// lookupCache returns the entry to restore for these keys, or (nil, nil) when there is none:
// either nothing matched, or the match had lost its blob to a prune, in which case the dangling
// entry is dropped on the way out.
-func (h *Handler) lookupCache(db *bolthold.Store, repo string, keys []string, version string) (*Cache, error) {
+func (h *Handler) lookupCache(db *cacheDB, repo string, keys []string, version string) (*Cache, error) {
cache, err := findCache(db, repo, keys, version)
if err != nil || cache == nil {
return nil, err
@@ -910,7 +942,7 @@ func (h *Handler) signedArtifactURL(cred JobCredential, cacheID uint64, exp time
}
// if not found, return (nil, nil) instead of an error.
-func findCache(db *bolthold.Store, repo string, keys []string, version string) (*Cache, error) {
+func findCache(db *cacheDB, repo string, keys []string, version string) (*Cache, error) {
cache := &Cache{}
for _, prefix := range keys {
// if a key in the list matches exactly, don't return partial matches
@@ -944,7 +976,7 @@ func findCache(db *bolthold.Store, repo string, keys []string, version string) (
// A completed entry is the one to restore, sorted by when it was written. An incomplete one is a
// reservation being uploaded to, sorted by when it was last written to, because the upload route
// touches UsedAt on every part.
-func findExactCache(db *bolthold.Store, repo, key, version string, complete bool) (*Cache, error) {
+func findExactCache(db *cacheDB, repo, key, version string, complete bool) (*Cache, error) {
sortBy := "UsedAt"
if complete {
sortBy = "CreatedAt"
@@ -965,7 +997,7 @@ func findExactCache(db *bolthold.Store, repo, key, version string, complete bool
return cache, nil
}
-func (h *Handler) insertCache(db *bolthold.Store, cache *Cache) error {
+func (h *Handler) insertCache(db *cacheDB, cache *Cache) error {
if h.s3 != nil {
var id [8]byte
_, _ = rand.Read(id[:])
@@ -1089,7 +1121,7 @@ func (h *Handler) gcCache() {
// evictForFreeSpace bounds the volume itself, so it also covers bytes the cache never
// accounted for.
-func (h *Handler) evictForFreeSpace(db *bolthold.Store) {
+func (h *Handler) evictForFreeSpace(db *cacheDB) {
free, err := h.freeDisk(h.dir)
if err != nil {
h.logger.Debugf("free disk check: %v", err) // unsupported platform, treat as unavailable rather than full
@@ -1109,14 +1141,14 @@ func (h *Handler) evictForFreeSpace(db *bolthold.Store) {
}
// evictIncomplete removes uploads that stopped part way, which are most likely broken.
-func (h *Handler) evictIncomplete(db *bolthold.Store) {
+func (h *Handler) evictIncomplete(db *cacheDB) {
h.sweep(db, bolthold.
Where("UsedAt").Lt(time.Now().Add(-uploadStallTimeout).Unix()).
And("Complete").Eq(false).
Index("UsedAt"))
}
-func (h *Handler) evictExpired(db *bolthold.Store) {
+func (h *Handler) evictExpired(db *cacheDB) {
if h.policy.Retention <= 0 {
return
}
@@ -1127,7 +1159,7 @@ func (h *Handler) evictExpired(db *bolthold.Store) {
// evictSuperseded removes entries a newer one with the same key and version replaced. The
// aggregation includes Repo so two repos sharing a (key, version) do not evict each other.
-func (h *Handler) evictSuperseded(db *bolthold.Store) {
+func (h *Handler) evictSuperseded(db *cacheDB) {
results, err := db.FindAggregate(&Cache{}, bolthold.Where("Complete").Eq(true).Index("Complete"), "Repo", "Key", "Version")
if err != nil {
h.logger.Warnf("find aggregate caches: %v", err)
@@ -1152,7 +1184,7 @@ func (h *Handler) evictSuperseded(db *bolthold.Store) {
// evictOversized applies the per-repository limit, then the whole-store one. Only completed
// entries count, since only those carry a size measured at commit rather than claimed.
-func (h *Handler) evictOversized(db *bolthold.Store) {
+func (h *Handler) evictOversized(db *cacheDB) {
if h.policy.RepoSizeLimit > 0 {
byRepo := make(map[string][]*Cache)
for _, cache := range h.completedByUse(db) {
@@ -1167,7 +1199,7 @@ func (h *Handler) evictOversized(db *bolthold.Store) {
// evictTotal caps the store as a whole. It re-queries because the per-repo pass may have
// deleted rows an earlier result still holds.
-func (h *Handler) evictTotal(db *bolthold.Store) {
+func (h *Handler) evictTotal(db *cacheDB) {
if h.policy.SizeLimit <= 0 {
return
}
@@ -1175,7 +1207,7 @@ func (h *Handler) evictTotal(db *bolthold.Store) {
}
// evictRepo reclaims space when a commit pushes a repo over, rather than at the next sweep.
-func (h *Handler) evictRepo(db *bolthold.Store, repo string) {
+func (h *Handler) evictRepo(db *cacheDB, repo string) {
if h.policy.RepoSizeLimit <= 0 {
return
}
@@ -1183,7 +1215,7 @@ func (h *Handler) evictRepo(db *bolthold.Store, repo string) {
}
// evictTo deletes until caches fit limit. caches must be ordered by UsedAt ascending.
-func (h *Handler) evictTo(db *bolthold.Store, caches []*Cache, limit int64, scope string) {
+func (h *Handler) evictTo(db *cacheDB, caches []*Cache, limit int64, scope string) {
// An entry bigger than the limit never fits, so it goes on its own account instead of
// dragging every neighbour out first and then following them next sweep.
fits := caches[:0]
@@ -1230,7 +1262,7 @@ func touchNeeded(cache *Cache) bool {
// touch stamps UsedAt through the caller's store, a bolt write on the read path. It cannot
// go through touchCache, which opens its own store and would block on the exclusive lock
// for as long as the caller holds one.
-func (h *Handler) touch(db *bolthold.Store, cache *Cache) {
+func (h *Handler) touch(db *cacheDB, cache *Cache) {
if !touchNeeded(cache) {
return
}
@@ -1240,13 +1272,13 @@ func (h *Handler) touch(db *bolthold.Store, cache *Cache) {
}
}
-func (h *Handler) sweep(db *bolthold.Store, query *bolthold.Query) {
+func (h *Handler) sweep(db *cacheDB, query *bolthold.Query) {
for _, cache := range h.caches(db, query) {
h.deleteCache(db, cache)
}
}
-func (h *Handler) caches(db *bolthold.Store, query *bolthold.Query) []*Cache {
+func (h *Handler) caches(db *cacheDB, query *bolthold.Query) []*Cache {
var caches []*Cache
if err := db.Find(&caches, query); err != nil {
h.logger.Warnf("find caches: %v", err)
@@ -1256,14 +1288,14 @@ func (h *Handler) caches(db *bolthold.Store, query *bolthold.Query) []*Cache {
// cachesByUse returns matches least recently accessed first, sorting here rather than with
// bolthold's SortBy, which reflects over every field it compares.
-func (h *Handler) cachesByUse(db *bolthold.Store, query *bolthold.Query) []*Cache {
+func (h *Handler) cachesByUse(db *cacheDB, query *bolthold.Query) []*Cache {
caches := h.caches(db, query)
slices.SortFunc(caches, func(a, b *Cache) int { return cmp.Compare(a.UsedAt, b.UsedAt) })
return caches
}
// completedByUse returns every entry the size limits count, least recently accessed first.
-func (h *Handler) completedByUse(db *bolthold.Store) []*Cache {
+func (h *Handler) completedByUse(db *cacheDB) []*Cache {
return h.cachesByUse(db, bolthold.Where("Complete").Eq(true).Index("Complete"))
}
@@ -1277,7 +1309,7 @@ func totalSize(caches []*Cache) int64 {
// deleteCache drops an entry and its bytes, reporting whether it went fully. The blob goes
// first, so a failed unlink leaves the row for the next sweep instead of orphaning bytes.
-func (h *Handler) deleteCache(db *bolthold.Store, cache *Cache) bool {
+func (h *Handler) deleteCache(db *cacheDB, cache *Cache) bool {
storage := h.storage
if h.s3 != nil && !cache.Complete {
storage = h.s3.Storage // an upload not yet committed has nothing in the bucket
diff --git a/act/artifactcache/handler_test.go b/act/artifactcache/handler_test.go
index 6af08431..b95730f4 100644
--- a/act/artifactcache/handler_test.go
+++ b/act/artifactcache/handler_test.go
@@ -17,12 +17,14 @@ import (
"path/filepath"
"strings"
"testing"
+ "testing/synctest"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/timshannon/bolthold"
"go.etcd.io/bbolt"
+ berrors "go.etcd.io/bbolt/errors"
)
// testToken is registered with the cache server in every test that needs to
@@ -114,6 +116,26 @@ func TestHandler(t *testing.T) {
base := fmt.Sprintf("%s%s", handler.ExternalURL(), apiPath)
+ t.Run("database open recovers from failure, times out while held and close is idempotent", func(t *testing.T) {
+ dir := t.TempDir()
+ synctest.Test(t, func(t *testing.T) {
+ handler := &Handler{dir: dir, dbLock: make(chan struct{}, 1)}
+ require.NoError(t, os.Mkdir(filepath.Join(handler.dir, "bolt.db"), 0o755))
+ _, err := handler.openDB()
+ require.Error(t, err)
+ require.NoError(t, os.Remove(filepath.Join(handler.dir, "bolt.db")))
+ first, err := handler.openDB()
+ require.NoError(t, err)
+ _, err = handler.openDB()
+ require.ErrorIs(t, err, berrors.ErrTimeout)
+ require.NoError(t, first.Close())
+ require.NoError(t, first.Close())
+ second, err := handler.openDB()
+ require.NoError(t, err)
+ require.NoError(t, second.Close())
+ })
+ })
+
defer func() {
t.Run("inpect db", func(t *testing.T) {
db, err := handler.openDB()
diff --git a/act/artifactcache/s3_storage.go b/act/artifactcache/s3_storage.go
index 1824b583..9df7f7df 100644
--- a/act/artifactcache/s3_storage.go
+++ b/act/artifactcache/s3_storage.go
@@ -24,6 +24,10 @@ import (
"strconv"
"strings"
"time"
+
+ "gitea.com/gitea/runner/act/common"
+
+ "golang.org/x/sync/errgroup"
)
// S3Options configures the bucket that stores completed cache entries.
@@ -47,7 +51,7 @@ type cacheStorage interface {
Remove(uint64) error
}
-const s3PartSize = 5 << 30 // the largest part S3 accepts
+const s3PartSize = 64 << 20
// s3Storage stages uploads on disk and keeps completed entries and their metadata in a bucket.
type s3Storage struct {
@@ -69,6 +73,7 @@ func newS3Storage(staging *Storage, opts S3Options) (*s3Storage, error) {
}
transport, _ := http.DefaultTransport.(*http.Transport)
transport = transport.Clone()
+ transport.MaxIdleConnsPerHost = 32 // 4 parts each for the 8 jobs a runner commonly runs
transport.ResponseHeaderTimeout = time.Minute // a stalled endpoint must not hold the store forever
pathStyle := !strings.HasSuffix(endpoint.Hostname(), ".amazonaws.com")
if opts.PathStyle != nil {
@@ -209,26 +214,43 @@ func (s *s3Storage) Commit(id uint64, size int64) (int64, error) {
return written, nil
}
-// upload always goes in parts, the only way S3 takes an object over 5 GiB.
-func (s *s3Storage) upload(ctx context.Context, key string, file io.ReaderAt, size int64) error {
+func (s *s3Storage) upload(ctx context.Context, key string, file io.ReaderAt, size int64) (uploadErr error) {
+ if size <= s3PartSize {
+ _, err := s.call(ctx, http.MethodPut, key, nil, io.NewSectionReader(file, 0, size), nil)
+ return err
+ }
var created struct {
UploadID string `xml:"UploadId"`
}
if _, err := s.call(ctx, http.MethodPost, key, url.Values{"uploads": {""}}, nil, &created); err != nil {
return err
}
+ defer func() {
+ if uploadErr != nil && created.UploadID != "" {
+ abortCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), time.Minute)
+ defer cancel()
+ _, _ = s.call(abortCtx, http.MethodDelete, key, url.Values{"uploadId": {created.UploadID}}, nil, nil)
+ }
+ }()
var completion struct {
XMLName xml.Name `xml:"CompleteMultipartUpload"`
Parts []s3Part `xml:"Part"`
}
- for offset := int64(0); offset == 0 || offset < size; offset += s3PartSize {
- number, length := len(completion.Parts)+1, min(s3PartSize, size-offset)
- query := url.Values{"partNumber": {strconv.Itoa(number)}, "uploadId": {created.UploadID}}
- header, err := s.call(ctx, http.MethodPut, key, query, io.NewSectionReader(file, offset, length), nil)
- if err != nil {
+ partSize := max(int64(s3PartSize), (size-1)/10000+1)
+ completion.Parts = make([]s3Part, (size-1)/partSize+1)
+ workers, workerCtx := errgroup.WithContext(ctx)
+ workers.SetLimit(4)
+ for part := range completion.Parts {
+ workers.Go(func() (err error) {
+ defer common.RecoverPanic(&err)
+ offset := int64(part) * partSize
+ header, err := s.call(workerCtx, http.MethodPut, key, url.Values{"partNumber": {strconv.Itoa(part + 1)}, "uploadId": {created.UploadID}}, io.NewSectionReader(file, offset, min(partSize, size-offset)), nil)
+ completion.Parts[part] = s3Part{part + 1, header.Get("ETag")}
return err
- }
- completion.Parts = append(completion.Parts, s3Part{number, header.Get("ETag")})
+ })
+ }
+ if err := workers.Wait(); err != nil {
+ return err
}
body, err := xml.Marshal(completion)
if err != nil {
diff --git a/act/artifactcache/s3_storage_test.go b/act/artifactcache/s3_storage_test.go
index dd60323f..c614ce76 100644
--- a/act/artifactcache/s3_storage_test.go
+++ b/act/artifactcache/s3_storage_test.go
@@ -4,8 +4,12 @@
package artifactcache
import (
+ "encoding/xml"
+ "io"
"net/http"
"net/http/httptest"
+ "strings"
+ "sync/atomic"
"testing"
"time"
@@ -18,7 +22,14 @@ import (
func TestHandler_S3SharesEntriesAndDropsExpiredOnes(t *testing.T) {
backend := s3mem.New()
require.NoError(t, backend.CreateBucket("cache"))
- server := httptest.NewServer(gofakes3.New(backend).Server())
+ var uploads atomic.Int32
+ handler := gofakes3.New(backend).Server()
+ server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+ if strings.Contains(request.URL.Path, "/blobs/") && (request.Method == http.MethodPut || request.Method == http.MethodPost) {
+ uploads.Add(1)
+ }
+ handler.ServeHTTP(writer, request)
+ }))
t.Cleanup(server.Close)
opts := Options{S3: &S3Options{Endpoint: server.URL, Bucket: "cache", Prefix: "runners", AccessKey: "key", SecretKey: "secret"}}
saver, restorer := startTestHandler(t, opts), startTestHandler(t, opts)
@@ -26,6 +37,7 @@ func TestHandler_S3SharesEntriesAndDropsExpiredOnes(t *testing.T) {
finalized, _ := saveV2(t, saver, "deps", "v1", []byte("archive"))
require.Equal(t, true, finalized["ok"])
+ assert.EqualValues(t, 1, uploads.Load())
require.NoError(t, restorer.importMetadata(t.Context()))
found := v2Call(t, restorer, testClient, "GetCacheEntryDownloadURL", entry)
require.Equal(t, true, found["ok"])
@@ -37,6 +49,51 @@ func TestHandler_S3SharesEntriesAndDropsExpiredOnes(t *testing.T) {
assert.Equal(t, true, v2Call(t, restorer, testClient, "CreateCacheEntry", entry)["ok"])
}
+type s3TestTransport func(*http.Request) (*http.Response, error)
+
+func (transport s3TestTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+ return transport(req)
+}
+
+func TestS3StorageUploadsLargeObjectsInPartsAndAbortsOnFailure(t *testing.T) {
+ for _, failingPart := range []string{"", "2"} {
+ var parts, aborts atomic.Int32
+ var completion struct {
+ Parts []s3Part `xml:"Part"`
+ }
+ storage := &s3Storage{bucketURL: httptest.NewRequest(http.MethodGet, "http://cache.local/bucket", nil).URL}
+ storage.client = &http.Client{Transport: s3TestTransport(func(req *http.Request) (*http.Response, error) {
+ query, response := req.URL.Query(), &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: http.NoBody}
+ switch {
+ case query.Has("uploads"):
+ response.Body = io.NopCloser(strings.NewReader("upload"))
+ case req.Method == http.MethodPut:
+ parts.Add(1)
+ response.Header.Set("ETag", "etag-"+query.Get("partNumber"))
+ if query.Get("partNumber") == failingPart {
+ response.StatusCode, response.Body = http.StatusBadRequest, io.NopCloser(strings.NewReader("InvalidPart"))
+ }
+ case req.Method == http.MethodPost:
+ require.NoError(t, xml.NewDecoder(req.Body).Decode(&completion))
+ response.Body = io.NopCloser(strings.NewReader(""))
+ case req.Method == http.MethodDelete:
+ aborts.Add(1)
+ }
+ return response, nil
+ })}
+ err := storage.upload(t.Context(), "blobs/1", strings.NewReader(""), 2*int64(s3PartSize)+1)
+ assert.EqualValues(t, 3, parts.Load())
+ if failingPart == "" {
+ require.NoError(t, err)
+ assert.Equal(t, []s3Part{{1, "etag-1"}, {2, "etag-2"}, {3, "etag-3"}}, completion.Parts)
+ assert.Zero(t, aborts.Load())
+ } else {
+ require.ErrorContains(t, err, "InvalidPart")
+ assert.EqualValues(t, 1, aborts.Load())
+ }
+ }
+}
+
func TestS3StorageSignsLikeAWS(t *testing.T) {
storage := &s3Storage{opts: S3Options{Region: "us-east-1", AccessKey: "AKIAIOSFODNN7EXAMPLE", SecretKey: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"}}
req := httptest.NewRequest(http.MethodGet, "https://examplebucket.s3.amazonaws.com/?max-keys=2&prefix=J", nil)
diff --git a/act/common/executor.go b/act/common/executor.go
index 26a2405f..045e27b5 100644
--- a/act/common/executor.go
+++ b/act/common/executor.go
@@ -61,6 +61,13 @@ func NewErrorExecutor(err error) Executor {
}
}
+func RecoverPanic(err *error) { // defer directly in workers to keep stacks in host logs
+ if recovered := recover(); recovered != nil {
+ logging.Errorf("panic: %v\n%s", recovered, debug.Stack())
+ *err = fmt.Errorf("panic: %v", recovered)
+ }
+}
+
// NewParallelExecutor creates a new executor from a parallel of other executors
func NewParallelExecutor(parallel int, executors ...Executor) Executor {
if len(executors) == 0 {
@@ -91,12 +98,7 @@ func NewParallelExecutor(parallel int, executors ...Executor) Executor {
// goroutine which would leave the runner process hung.
// https://gitea.com/gitea/runner/issues/371
errs <- func() (err error) {
- defer func() {
- if r := recover(); r != nil {
- logging.Errorf("panic in executor: %v\n%s", r, debug.Stack())
- err = fmt.Errorf("panic: %v", r)
- }
- }()
+ defer RecoverPanic(&err)
return executor(ctx)
}()
}
diff --git a/act/common/git/fetch_action.go b/act/common/git/fetch_action.go
index 858cab3f..eef43669 100644
--- a/act/common/git/fetch_action.go
+++ b/act/common/git/fetch_action.go
@@ -16,14 +16,25 @@ import (
"net/url"
"os"
"path/filepath"
+ "reflect"
"runtime"
"strconv"
"strings"
+ "sync"
"time"
"gitea.com/gitea/runner/act/common"
)
+var actionFetches sync.Map
+
+type actionFetch struct {
+ done chan struct{}
+ dir, sha string
+ err error
+ retry bool // the leader was cancelled or panicked, so followers fetch themselves
+}
+
var archiveRetryDelay = func(retryAfter string) time.Duration { // as actions/runner backs off
if seconds, err := strconv.Atoi(retryAfter); err == nil {
return min(max(time.Duration(seconds)*time.Second, 10*time.Second), 10*time.Minute)
@@ -32,6 +43,36 @@ var archiveRetryDelay = func(retryAfter string) time.Duration { // as actions/ru
}
func FetchAction(ctx context.Context, input NewGitCloneExecutorInput) (string, string, error) { // input.Dir is one clone URL and ref's, holding a directory per commit
+ if !reflect.ValueOf(input).Comparable() {
+ return fetchAction(ctx, input)
+ }
+ for ctx.Err() == nil {
+ call := &actionFetch{done: make(chan struct{}), retry: true}
+ previous, loaded := actionFetches.LoadOrStore(input, call)
+ if !loaded {
+ func() {
+ defer func() { actionFetches.Delete(input); close(call.done) }()
+ call.dir, call.sha, call.err = fetchAction(ctx, input)
+ call.retry = ctx.Err() != nil
+ }()
+ if call.retry {
+ continue
+ }
+ return call.dir, call.sha, call.err
+ }
+ call, _ = previous.(*actionFetch)
+ select {
+ case <-ctx.Done():
+ case <-call.done:
+ if !call.retry && ctx.Err() == nil {
+ return call.dir, call.sha, call.err
+ }
+ }
+ }
+ return "", "", ctx.Err()
+}
+
+func fetchAction(ctx context.Context, input NewGitCloneExecutorInput) (string, string, error) {
unlock, err := AcquireCloneLock(ctx, input.Dir)
if err != nil {
return "", "", err
diff --git a/act/common/git/fetch_action_test.go b/act/common/git/fetch_action_test.go
index c71987b9..fa4247c5 100644
--- a/act/common/git/fetch_action_test.go
+++ b/act/common/git/fetch_action_test.go
@@ -6,6 +6,7 @@ package git
import (
"archive/tar"
"bytes"
+ "context"
"encoding/base64"
"fmt"
"net/http"
@@ -16,9 +17,12 @@ import (
"path/filepath"
"strings"
"sync"
+ "sync/atomic"
"testing"
"time"
+ "gitea.com/gitea/runner/act/common"
+
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -102,6 +106,75 @@ func TestFetchAction(t *testing.T) {
return NewGitCloneExecutorInput{URL: server.URL + "/org/action", Ref: ref, Dir: t.TempDir(), Token: "secret"}
}
+ t.Run("shares identical fetches and isolates cancellation, panics and differing inputs", func(t *testing.T) {
+ expectedSha := gitRevParse(t, remote, "main")
+ for _, scenario := range []string{"shared", "leader cancelled", "leader panics", "different token"} {
+ t.Run(scenario, func(t *testing.T) {
+ serve("prefixed", 0)
+ input := newInput("main")
+ started, release := make(chan struct{}), make(chan struct{})
+ releaseFetches := sync.OnceFunc(func() { close(release) })
+ t.Cleanup(releaseFetches)
+ var once sync.Once
+ var panicked atomic.Bool
+ transport := actionFetchTransport(func(request *http.Request) (*http.Response, error) {
+ once.Do(func() { close(started) })
+ select {
+ case <-request.Context().Done():
+ return nil, request.Context().Err()
+ case <-release:
+ }
+ if scenario == "leader panics" && panicked.CompareAndSwap(false, true) {
+ panic("broken archive")
+ }
+ return http.DefaultTransport.RoundTrip(request)
+ })
+ input.Transport = &transport
+ leaderCtx, cancelLeader := context.WithCancel(t.Context())
+ defer cancelLeader()
+ waiting := &actionFetchContext{Context: t.Context(), waiting: make(chan struct{})}
+ followerInput := input
+ expectedRequests := 1
+ if scenario == "different token" {
+ followerInput.Token, expectedRequests = "other", 2
+ }
+ leaderDone, followerDone := make(chan error, 1), make(chan error, 1)
+ fetch := func(ctx context.Context, input NewGitCloneExecutorInput, done chan<- error) {
+ var err error
+ defer func() { done <- err }()
+ defer common.RecoverPanic(&err)
+ var tree, sha string
+ if tree, sha, err = FetchAction(ctx, input); err == nil {
+ assert.Equal(t, expectedSha, sha)
+ assert.Equal(t, filepath.Join(input.Dir, sha), tree)
+ }
+ }
+ go fetch(leaderCtx, input, leaderDone)
+ <-started
+ go fetch(waiting, followerInput, followerDone)
+ <-waiting.waiting
+ if scenario == "leader cancelled" {
+ cancelLeader()
+ require.ErrorIs(t, <-leaderDone, context.Canceled)
+ }
+ releaseFetches()
+ switch scenario {
+ case "leader cancelled":
+ case "leader panics":
+ require.ErrorContains(t, <-leaderDone, "panic: broken archive")
+ default:
+ require.NoError(t, <-leaderDone)
+ }
+ require.NoError(t, <-followerDone)
+ _, headers := served()
+ assert.Len(t, headers, expectedRequests)
+ if scenario == "different token" {
+ assert.NotEqual(t, headers[0].Get("Authorization"), headers[1].Get("Authorization"))
+ }
+ })
+ }
+ })
+
t.Run("downloads, revalidates and follows a moved tag", func(t *testing.T) {
serve("prefixed", 0)
input := newInput("v1")
@@ -200,6 +273,23 @@ func TestFetchAction(t *testing.T) {
})
}
+type actionFetchTransport func(*http.Request) (*http.Response, error)
+
+func (transport actionFetchTransport) RoundTrip(request *http.Request) (*http.Response, error) {
+ return transport(request)
+}
+
+type actionFetchContext struct {
+ context.Context
+ waiting chan struct{}
+ once sync.Once
+}
+
+func (ctx *actionFetchContext) Done() <-chan struct{} {
+ ctx.once.Do(func() { close(ctx.waiting) })
+ return ctx.Context.Done()
+}
+
func TestExtractTreeConfinesEntries(t *testing.T) {
outside := t.TempDir()
for name, entries := range map[string][]*tar.Header{
diff --git a/act/container/docker_images.go b/act/container/docker_images.go
index 1437bced..c4c7d26b 100644
--- a/act/container/docker_images.go
+++ b/act/container/docker_images.go
@@ -58,18 +58,14 @@ func RemoveImage(ctx context.Context, imageName string, force, pruneChildren boo
}
defer cli.Close()
- if _, err := cli.ImageInspect(ctx, imageName); cerrdefs.IsNotFound(err) {
+ if _, err = cli.ImageRemove(ctx, imageName, client.ImageRemoveOptions{
+ Force: force,
+ PruneChildren: pruneChildren,
+ }); cerrdefs.IsNotFound(err) {
return false, nil
} else if err != nil {
return false, err
}
- if _, err = cli.ImageRemove(ctx, imageName, client.ImageRemoveOptions{
- Force: force,
- PruneChildren: pruneChildren,
- }); err != nil {
- return false, err
- }
-
return true, nil
}
diff --git a/act/container/docker_images_test.go b/act/container/docker_images_test.go
index 233b5017..8942fd46 100644
--- a/act/container/docker_images_test.go
+++ b/act/container/docker_images_test.go
@@ -66,4 +66,11 @@ func TestImageExistsLocally(t *testing.T) {
wrongPlatform, err := ImageExistsLocally(ctx, amd64Ref, "linux/arm64")
assert.NoError(t, err) //nolint:testifylint // pre-existing issue from nektos/act
assert.False(t, wrongPlatform)
+
+ removed, err := RemoveImage(ctx, amd64Ref, false, false)
+ require.NoError(t, err)
+ assert.True(t, removed)
+ removed, err = RemoveImage(ctx, amd64Ref, false, false)
+ require.NoError(t, err)
+ assert.False(t, removed)
}
diff --git a/act/container/docker_proxy.go b/act/container/docker_proxy.go
index e7b39b69..7e2cdc25 100644
--- a/act/container/docker_proxy.go
+++ b/act/container/docker_proxy.go
@@ -48,6 +48,13 @@ var (
rawStreamPath = regexp.MustCompile(`^(/v[0-9.]+)?/(containers/[^/]+/attach|exec/[^/]+/start)$`)
)
+var dockerProxyDirectories sync.Map
+
+type dockerProxyDirectory struct {
+ dir, daemonDir string
+ socket os.FileInfo
+}
+
func NewDockerProxy(ctx context.Context, job string) *DockerProxy {
if host := os.Getenv("DOCKER_HOST"); runtime.GOOS != "linux" || host != "" && !strings.HasPrefix(host, "unix://") {
return nil
@@ -63,15 +70,24 @@ func NewDockerProxy(ctx context.Context, job string) *DockerProxy {
if !ok {
return nil
}
- if info, err := os.Stat(daemonSocket); err != nil || info.Mode()&os.ModeSocket == 0 {
+ socket, err := os.Stat(daemonSocket)
+ if err != nil || socket.Mode()&os.ModeSocket == 0 {
return nil
}
- dir, daemonDir := runnerContainerWorkdir(probeCtx, cli)
- seen := daemonDir != ""
+ workdir, _ := os.Getwd()
+ key := [3]string{cli.DaemonHost(), workdir, os.TempDir()}
+ cached, _ := dockerProxyDirectories.Load(key)
+ directory, seen := cached.(dockerProxyDirectory)
+ seen = seen && os.SameFile(directory.socket, socket) // daemon replacement invalidates cached paths
+ dir, daemonDir := directory.dir, directory.daemonDir
if !seen {
- if dir, err = filepath.Abs(os.TempDir()); err == nil {
- daemonDir = dir
- seen, err = daemonSeesDir(probeCtx, cli, dir, daemonDir)
+ dir, daemonDir = runnerContainerWorkdir(probeCtx, cli)
+ seen = daemonDir != ""
+ if !seen {
+ if dir, err = filepath.Abs(os.TempDir()); err == nil {
+ daemonDir = dir
+ seen, err = daemonSeesDir(probeCtx, cli, dir, daemonDir)
+ }
}
}
if err != nil {
@@ -87,9 +103,11 @@ func NewDockerProxy(ctx context.Context, job string) *DockerProxy {
}
proxy, err := StartDockerProxy(daemonSocket, dir, job)
if err != nil {
+ dockerProxyDirectories.Delete(key)
common.Logger(ctx).Warnf("docker proxy not started, the job gets the daemon socket directly: %v", err)
return nil
}
+ dockerProxyDirectories.Store(key, dockerProxyDirectory{dir, daemonDir, socket})
proxy.Socket = daemonDir + strings.TrimPrefix(proxy.Socket, dir)
return proxy
}
diff --git a/act/container/docker_proxy_test.go b/act/container/docker_proxy_test.go
index 18cb2812..36a1ae64 100644
--- a/act/container/docker_proxy_test.go
+++ b/act/container/docker_proxy_test.go
@@ -20,6 +20,7 @@ import (
"runtime"
"strconv"
"strings"
+ "sync/atomic"
"testing"
"time"
@@ -68,6 +69,7 @@ func TestDockerProxy(t *testing.T) {
bodies := make(chan []byte, 8)
createStarted, releaseCreate := make(chan struct{}), make(chan struct{})
streamsDone := make(chan struct{}, 2)
+ var probes atomic.Int32
daemonSocket := filepath.Join(shortTempDir(t), "d.sock")
listener, err := net.Listen("unix", daemonSocket)
require.NoError(t, err)
@@ -75,12 +77,28 @@ func TestDockerProxy(t *testing.T) {
switch {
case strings.HasSuffix(r.URL.Path, "/create"):
body, _ := io.ReadAll(r.Body)
+ if bytes.Contains(body, []byte(`"Image":"probe-image"`)) {
+ _, _ = w.Write([]byte(`{"Id":"probe"}`))
+ return
+ }
if r.URL.RawQuery == "wait" {
close(createStarted)
<-releaseCreate
}
bodies <- body
w.WriteHeader(http.StatusCreated)
+ case strings.HasSuffix(r.URL.Path, "/images/json"):
+ if probes.Add(1) == 1 {
+ http.Error(w, "probe unavailable", http.StatusServiceUnavailable)
+ return
+ }
+ _, _ = w.Write([]byte(`[{"Id":"probe-image"}]`))
+ case strings.HasSuffix(r.URL.Path, "/json"):
+ http.NotFound(w, r)
+ case strings.HasSuffix(r.URL.Path, "/archive"):
+ w.Header().Set("X-Docker-Container-Path-Stat", "e30=")
+ case r.Method == http.MethodDelete:
+ w.WriteHeader(http.StatusNoContent)
case r.URL.Path == "/exec/detached/start" || r.URL.Path == "/exec/error/start":
if r.URL.Path == "/exec/error/start" {
w.Header().Set("Content-Type", "application/vnd.docker.raw-stream")
@@ -211,6 +229,31 @@ func TestDockerProxy(t *testing.T) {
assert.Equal(t, "OK GET /v1.47/_ping", string(body))
})
+ t.Run("discovery caches only successes", func(t *testing.T) {
+ if runtime.GOOS != "linux" {
+ t.Skip("automatic proxy discovery requires Linux")
+ }
+ t.Setenv("DOCKER_HOST", "unix://"+daemonSocket)
+ t.Setenv("TMPDIR", shortTempDir(t))
+ require.Nil(t, NewDockerProxy(t.Context(), "failed"))
+ for _, job := range []string{"first", "second"} {
+ proxy := NewDockerProxy(t.Context(), job)
+ require.NotNil(t, proxy)
+ require.NoError(t, proxy.Close(context.Background()))
+ }
+ assert.EqualValues(t, 2, probes.Load())
+ dockerProxyDirectories.Range(func(key, value any) bool {
+ directory, _ := value.(dockerProxyDirectory)
+ directory.socket, _ = os.Stat(t.TempDir())
+ dockerProxyDirectories.Store(key, directory)
+ return true
+ })
+ proxy := NewDockerProxy(t.Context(), "replaced daemon")
+ require.NotNil(t, proxy)
+ require.NoError(t, proxy.Close(context.Background()))
+ assert.EqualValues(t, 3, probes.Load())
+ })
+
t.Run("tunnels raw streams through stdin EOF", func(t *testing.T) {
for _, stream := range []struct{ path, upgrade string }{
{path: "/v1.47/exec/abc/start"},
diff --git a/act/container/docker_run.go b/act/container/docker_run.go
index 4681d06b..508b5fb5 100644
--- a/act/container/docker_run.go
+++ b/act/container/docker_run.go
@@ -476,22 +476,23 @@ func (cr *containerReference) find() common.Executor {
}
cr.id = ""
}
- containers, err := cr.cli.ContainerList(ctx, client.ContainerListOptions{
- All: true,
- })
+ result, err := cr.cli.ContainerInspect(ctx, cr.input.Name, client.ContainerInspectOptions{})
+ if err == nil && result.Container.Name == "/"+cr.input.Name {
+ cr.id = result.Container.ID
+ return nil
+ } else if cerrdefs.IsNotFound(err) {
+ return nil
+ }
+ containers, err := cr.cli.ContainerList(ctx, client.ContainerListOptions{All: true}) // inspect also matches ids, and podman fails it with a 500 for a container removed mid-request
if err != nil {
return fmt.Errorf("failed to list containers: %w", err)
}
-
- for _, c := range containers.Items {
- for _, name := range c.Names {
- if name[1:] == cr.input.Name {
- cr.id = c.ID
- return nil
- }
+ for _, candidate := range containers.Items {
+ if slices.Contains(candidate.Names, "/"+cr.input.Name) {
+ cr.id = candidate.ID
+ break
}
}
-
return nil
}
}
diff --git a/act/container/docker_run_test.go b/act/container/docker_run_test.go
index 4ba08d8d..7ac89c44 100644
--- a/act/container/docker_run_test.go
+++ b/act/container/docker_run_test.go
@@ -13,9 +13,13 @@ import (
"errors"
"io"
"net"
+ "net/http"
+ "net/http/httptest"
"os"
"path/filepath"
"strings"
+ "sync"
+ "sync/atomic"
"testing"
"gitea.com/gitea/runner/act/common"
@@ -408,13 +412,12 @@ func TestFindRevalidatesStaleID(t *testing.T) {
client := &mockDockerClient{}
return &containerReference{id: id, cli: client, input: &NewContainerInput{Name: "job-1"}}, client
}
- listOpts := mobyclient.ContainerListOptions{All: true}
inspectOpts := mobyclient.ContainerInspectOptions{}
t.Run("stale id cleared, name lookup empty", func(t *testing.T) {
cr, client := newCR("stale")
client.On("ContainerInspect", ctx, "stale", inspectOpts).Return(mobyclient.ContainerInspectResult{}, notFound)
- client.On("ContainerList", ctx, listOpts).Return(mobyclient.ContainerListResult{}, nil)
+ client.On("ContainerInspect", ctx, "job-1", inspectOpts).Return(mobyclient.ContainerInspectResult{}, notFound)
require.NoError(t, cr.find()(ctx))
assert.Empty(t, cr.id)
client.AssertExpectations(t)
@@ -423,8 +426,18 @@ func TestFindRevalidatesStaleID(t *testing.T) {
t.Run("stale id cleared, name lookup repopulates", func(t *testing.T) {
cr, client := newCR("stale")
client.On("ContainerInspect", ctx, "stale", inspectOpts).Return(mobyclient.ContainerInspectResult{}, notFound)
- client.On("ContainerList", ctx, listOpts).Return(mobyclient.ContainerListResult{Items: []container.Summary{
- {ID: "other", Names: []string{"/somebody-else"}},
+ client.On("ContainerInspect", ctx, "job-1", inspectOpts).
+ Return(mobyclient.ContainerInspectResult{Container: container.InspectResponse{ID: "fresh", Name: "/job-1"}}, nil)
+ require.NoError(t, cr.find()(ctx))
+ assert.Equal(t, "fresh", cr.id)
+ client.AssertExpectations(t)
+ })
+
+ t.Run("id match falls back to listing", func(t *testing.T) {
+ cr, client := newCR("")
+ client.On("ContainerInspect", ctx, "job-1", inspectOpts).
+ Return(mobyclient.ContainerInspectResult{Container: container.InspectResponse{ID: "job-1abc", Name: "/somebody-else"}}, nil)
+ client.On("ContainerList", ctx, mobyclient.ContainerListOptions{All: true}).Return(mobyclient.ContainerListResult{Items: []container.Summary{
{ID: "fresh", Names: []string{"/job-1"}},
}}, nil)
require.NoError(t, cr.find()(ctx))
@@ -448,20 +461,52 @@ func TestFindRevalidatesStaleID(t *testing.T) {
client.AssertExpectations(t)
})
- t.Run("list error propagates", func(t *testing.T) {
+ t.Run("name lookup error falls back to listing", func(t *testing.T) {
cr, client := newCR("")
- client.On("ContainerList", ctx, listOpts).Return(mobyclient.ContainerListResult{}, boom)
+ client.On("ContainerInspect", ctx, "job-1", inspectOpts).Return(mobyclient.ContainerInspectResult{}, boom)
+ client.On("ContainerList", ctx, mobyclient.ContainerListOptions{All: true}).Return(mobyclient.ContainerListResult{Items: []container.Summary{
+ {ID: "other", Names: []string{"/somebody-else"}},
+ {ID: "fresh", Names: []string{"/job-1"}},
+ }}, nil).Once()
+ require.NoError(t, cr.find()(ctx))
+ assert.Equal(t, "fresh", cr.id)
+ client.On("ContainerList", ctx, mobyclient.ContainerListOptions{All: true}).Return(mobyclient.ContainerListResult{}, boom)
+ cr.id = ""
require.ErrorIs(t, cr.find()(ctx), boom)
client.AssertExpectations(t)
})
}
+func TestRunnerArchCachedPerRun(t *testing.T) {
+ var infos atomic.Int32
+ daemon := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+ writer.Header().Set("Content-Type", "application/json")
+ writer.Header().Set("API-Version", "1.47")
+ if strings.HasSuffix(request.URL.Path, "/info") {
+ infos.Add(1)
+ _, _ = io.WriteString(writer, `{"Architecture":"aarch64"}`)
+ }
+ }))
+ t.Cleanup(daemon.Close)
+ t.Setenv("DOCKER_HOST", daemon.URL)
+ extensions := &LinuxContainerEnvironmentExtensions{}
+ run := WithRunnerArch(t.Context())
+ var lookups sync.WaitGroup
+ for range 4 {
+ lookups.Go(func() { assert.Equal(t, "ARM64", extensions.GetRunnerContext(run)["arch"]) })
+ }
+ lookups.Wait()
+ assert.Equal(t, "ARM64", extensions.GetRunnerContext(t.Context())["arch"])
+ assert.EqualValues(t, 2, infos.Load())
+}
+
// Every daemon entry point fails fast with a clear, container-named
// error when no live cr.id is known.
func TestRejectsMissingContainer(t *testing.T) {
ctx := t.Context()
client := &mockDockerClient{}
- client.On("ContainerList", ctx, mobyclient.ContainerListOptions{All: true}).Return(mobyclient.ContainerListResult{}, nil)
+ client.On("ContainerInspect", ctx, "job-1", mobyclient.ContainerInspectOptions{}).
+ Return(mobyclient.ContainerInspectResult{}, cerrdefs.ErrNotFound)
cr := &containerReference{cli: client, input: &NewContainerInput{Name: "job-1"}}
check := func(op string, err error) {
t.Helper()
@@ -498,10 +543,8 @@ func TestPublicCopyPipelineHandlesStaleID(t *testing.T) {
client := &mockDockerClient{}
client.On("ContainerInspect", ctx, "stale", mobyclient.ContainerInspectOptions{}).
Return(mobyclient.ContainerInspectResult{}, cerrdefs.ErrNotFound.WithMessage("gone"))
- client.On("ContainerList", ctx, mobyclient.ContainerListOptions{All: true}).
- Return(mobyclient.ContainerListResult{Items: []container.Summary{
- {ID: "fresh", Names: []string{"/job-1"}},
- }}, nil)
+ client.On("ContainerInspect", ctx, "job-1", mobyclient.ContainerInspectOptions{}).
+ Return(mobyclient.ContainerInspectResult{Container: container.InspectResponse{ID: "fresh", Name: "/job-1"}}, nil)
client.On("CopyToContainer", ctx, "fresh", mock.Anything).
Run(func(args mock.Arguments) {
opts := args.Get(2).(mobyclient.CopyToContainerOptions)
diff --git a/act/container/linux_container_environment_extensions.go b/act/container/linux_container_environment_extensions.go
index 11184437..1c21ab39 100644
--- a/act/container/linux_container_environment_extensions.go
+++ b/act/container/linux_container_environment_extensions.go
@@ -69,10 +69,41 @@ func (*LinuxContainerEnvironmentExtensions) JoinPathVariable(paths ...string) st
// DefaultToolCache is where the runner mounts the tool cache inside job containers.
const DefaultToolCache = "/opt/hostedtoolcache"
+type runnerArchKey struct{}
+
+func WithRunnerArch(ctx context.Context) context.Context { // shares one architecture lookup per run
+ if ctx.Value(runnerArchKey{}) != nil {
+ return ctx
+ }
+ return context.WithValue(ctx, runnerArchKey{}, &runnerArchCache{lock: make(chan struct{}, 1)})
+}
+
+type runnerArchCache struct {
+ lock chan struct{} // lets cancelled jobs stop waiting for the cache
+ arch string
+}
+
+func cachedRunnerArch(ctx context.Context) string {
+ cache, ok := ctx.Value(runnerArchKey{}).(*runnerArchCache)
+ if !ok {
+ return RunnerArch(ctx)
+ }
+ select {
+ case cache.lock <- struct{}{}:
+ defer func() { <-cache.lock }()
+ case <-ctx.Done():
+ return ""
+ }
+ if cache.arch == "" {
+ cache.arch = RunnerArch(ctx)
+ }
+ return cache.arch
+}
+
func (*LinuxContainerEnvironmentExtensions) GetRunnerContext(ctx context.Context) map[string]any {
return map[string]any{
"os": "Linux",
- "arch": RunnerArch(ctx),
+ "arch": cachedRunnerArch(ctx),
"temp": "/tmp",
"tool_cache": DefaultToolCache,
}
diff --git a/act/runner/expression.go b/act/runner/expression.go
index 2b6d0125..8eb6c329 100644
--- a/act/runner/expression.go
+++ b/act/runner/expression.go
@@ -49,7 +49,9 @@ func (rc *RunContext) NewExpressionEvaluatorWithEnv(ctx context.Context, env map
result := model.WorkflowCallResult{
Outputs: map[string]string{},
}
+ unlock := lockJob(job)
maps.Copy(result.Outputs, job.Outputs)
+ unlock()
workflowCallResult[jobName] = &result
}
}
diff --git a/act/runner/expression_test.go b/act/runner/expression_test.go
index a1ab7fc8..715f028a 100644
--- a/act/runner/expression_test.go
+++ b/act/runner/expression_test.go
@@ -6,6 +6,7 @@ package runner
import (
"strings"
+ "sync"
"testing"
"gitea.dev/actionslib/pkg/exprparser"
@@ -500,3 +501,22 @@ jobs:
assert.NotContains(t, a, "s3cr3t") // it reaches the container name, which no log masker covers
assert.NotEqual(t, a, b) // masking the name must not collapse two jobs onto one container
}
+
+func TestNewExpressionEvaluatorReadsJobOutputsWhileCombinationsWriteThem(t *testing.T) {
+ rc := createRunContext(t)
+ rc.caller = &caller{runContext: createRunContext(t)}
+ rc.ExprEval = rc.NewExpressionEvaluator(t.Context())
+ job := rc.Run.Job()
+ var writes sync.WaitGroup
+ writes.Go(func() {
+ for range 100 {
+ unlock := lockJob(job)
+ job.Outputs = map[string]string{"out": "value"}
+ unlock()
+ }
+ })
+ for range 100 {
+ rc.NewExpressionEvaluator(t.Context())
+ }
+ writes.Wait()
+}
diff --git a/act/runner/job_executor.go b/act/runner/job_executor.go
index e067c7b3..69accaa9 100644
--- a/act/runner/job_executor.go
+++ b/act/runner/job_executor.go
@@ -19,10 +19,12 @@ import (
"slices"
"strconv"
"strings"
+ "sync"
"time"
"unicode"
"gitea.com/gitea/runner/act/common"
+ "gitea.com/gitea/runner/act/common/logging"
"gitea.com/gitea/runner/act/container"
"gitea.dev/actionslib/pkg/exprparser"
@@ -95,6 +97,39 @@ func printPrepareActions(rc *RunContext, preparers []actionPreparer) common.Exec
rawLogger := common.Logger(ctx).With(rawOutputField, true)
rawLogger.Infof("Prepare all required actions")
+ fetchCtx, cancelFetches := context.WithCancel(ctx)
+ var fetches sync.WaitGroup
+ defer func() { cancelFetches(); fetches.Wait() }()
+ slots := make(chan struct{}, 4)
+ job := rc.topLevelRunContext()
+ if job.actionDownloads == nil {
+ job.actionDownloads = map[string]*actionDownload{}
+ }
+ for _, preparer := range preparers {
+ action, ok := preparer.(*stepActionRemote)
+ if !ok || strings.Contains(action.Step.Uses, "${{") || action.action != nil {
+ continue
+ }
+ github := action.getGithubContext(common.WithLogger(ctx, logging.New(logging.DiscardHandler)))
+ remote, err := newRemoteAction(action.Step.Uses, github, action.RunContext.Config.GitHubInstance)
+ if err != nil || action.skipsLocalCheckout(remote, github) {
+ continue
+ }
+ key := remote.downloadKey(action.RunContext.Config.DefaultActionURL())
+ if job.actionDownloads[key] != nil {
+ continue
+ }
+ download := &actionDownload{fetched: make(chan struct{})}
+ job.actionDownloads[key] = download
+ fetches.Go(func() {
+ defer close(download.fetched)
+ defer common.RecoverPanic(&download.err)
+ slots <- struct{}{}
+ defer func() { <-slots }()
+ download.dir, download.sha, download.err = action.fetchAction(fetchCtx, remote, github.Token)
+ })
+ }
+
for _, preparer := range preparers {
if err := preparer.prepareActionExecutor()(ctx); err != nil {
// No step has run yet, so the failure belongs to the job.
diff --git a/act/runner/job_executor_test.go b/act/runner/job_executor_test.go
index 8e0ed908..987538e8 100644
--- a/act/runner/job_executor_test.go
+++ b/act/runner/job_executor_test.go
@@ -14,13 +14,17 @@ import (
"io"
"net/http"
"net/http/httptest"
+ "path/filepath"
"slices"
"strconv"
"strings"
+ "sync/atomic"
"testing"
+ "testing/synctest"
"time"
"gitea.com/gitea/runner/act/common"
+ "gitea.com/gitea/runner/act/common/git"
"gitea.com/gitea/runner/act/common/logging"
"gitea.com/gitea/runner/act/common/logging/logtest"
"gitea.com/gitea/runner/act/container"
@@ -118,14 +122,13 @@ func (sfm *stepFactoryMock) newStep(model *model.Step, rc *RunContext) (step, er
type actionPreparerMock struct {
reference string
sha string
- err error
prepared int
}
func (apm *actionPreparerMock) prepareActionExecutor() common.Executor {
return func(context.Context) error {
apm.prepared++
- return apm.err
+ return nil
}
}
@@ -133,24 +136,85 @@ func (apm *actionPreparerMock) actionDownloadInfo() (string, string) {
return apm.reference, apm.sha
}
-func TestPrintPrepareActionsGolden(t *testing.T) {
- buf := &bytes.Buffer{}
- logger := logging.New(logging.NewLevelHandler(newJobLogHandler(buf, cyan), logging.LevelInfo))
- ctx := common.WithLogger(t.Context(), logger.With("job", "j1"))
+func TestPrintPrepareActionsFetchesConcurrently(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ buf := &bytes.Buffer{}
+ ctx := common.WithLogger(t.Context(), logging.New(logging.NewLevelHandler(newJobLogHandler(buf, cyan), logging.LevelInfo)).With("job", "j1"))
+ rc := newPrefetchRunContext(t)
+ started, release := make(chan string), make(chan struct{})
+ var fetched atomic.Int32
+ setFetch(t, func(_ctx context.Context, input git.NewGitCloneExecutorInput) (string, string, error) {
+ fetched.Add(1)
+ started <- input.Ref
+ <-release
+ return filepath.Join(input.Dir, "commit"), "sha-" + input.Ref, nil
+ })
+ preparers := append(newRemotePreparers(rc, "org/repo@v1", "org/repo/sub@v1", "org/repo@v2", "org/repo@v3", "org/repo@v4", "org/repo@v5"), &actionPreparerMock{})
+ finished := make(chan error)
+ go func() { finished <- printPrepareActions(rc, preparers)(ctx) }()
+ for range 4 {
+ <-started
+ }
+ synctest.Wait()
+ assert.EqualValues(t, 4, fetched.Load())
+ close(release)
+ <-started
+ require.NoError(t, <-finished)
+ assert.EqualValues(t, 5, fetched.Load())
+ lines := []string{"[j1] | Prepare all required actions"}
+ for _, ref := range []string{"v1", "v2", "v3", "v4", "v5"} {
+ lines = append(lines, fmt.Sprintf("[j1] | Download action repository 'org/repo@%s' (SHA:sha-%s)", ref, ref))
+ }
+ assert.Equal(t, strings.Join(append(lines, ""), "\n"), buf.String())
+ })
+}
- preparers := []actionPreparer{
- &actionPreparerMock{reference: "actions/checkout@v7", sha: "9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0"},
- // A step that downloads nothing, such as the checkout of the workflow's own repository.
- &actionPreparerMock{},
+func TestPrintPrepareActionsFailsJobAndCancelsLaterDownloads(t *testing.T) {
+ for name, panicFetch := range map[string]bool{"error": false, "panic": true} {
+ t.Run(name, func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ rc := newPrefetchRunContext(t)
+ setFetch(t, func(ctx context.Context, input git.NewGitCloneExecutorInput) (string, string, error) {
+ if input.Ref == "v1" {
+ if panicFetch {
+ panic("broken archive")
+ }
+ return "", "", assert.AnError
+ }
+ <-ctx.Done()
+ return "", "", ctx.Err()
+ })
+ ctx := common.WithJobErrorContainer(common.WithLogger(t.Context(), logging.New(logging.DiscardHandler)))
+ remaining := &actionPreparerMock{}
+ err := printPrepareActions(rc, append(newRemotePreparers(rc, "org/repo@v1", "org/repo@v2"), remaining))(ctx)
+ if panicFetch {
+ require.ErrorContains(t, err, "panic: broken archive")
+ } else {
+ require.ErrorIs(t, err, assert.AnError)
+ }
+ require.ErrorIs(t, err, common.JobError(ctx))
+ assert.True(t, rc.jobFailed)
+ assert.Zero(t, remaining.prepared)
+ })
+ })
}
- require.NoError(t, printPrepareActions(&RunContext{}, preparers)(ctx))
+}
- want := strings.Join([]string{
- "[j1] | Prepare all required actions",
- "[j1] | Download action repository 'actions/checkout@v7' (SHA:9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0)",
- "",
- }, "\n")
- assert.Equal(t, want, buf.String())
+func newPrefetchRunContext(t *testing.T) *RunContext {
+ return &RunContext{
+ Config: &Config{ActionCacheDir: t.TempDir(), PresetGitHubContext: &model.GithubContext{}},
+ Run: &model.Run{JobID: "1", Workflow: &model.Workflow{Jobs: map[string]*model.Job{"1": {}}}},
+ }
+}
+
+func newRemotePreparers(rc *RunContext, uses ...string) []actionPreparer {
+ preparers := make([]actionPreparer, 0, len(uses))
+ for _, use := range uses {
+ preparers = append(preparers, &stepActionRemote{Step: &model.Step{Uses: use}, RunContext: rc, readAction: func(_ctx context.Context, _step *model.Step, _actionDir, _actionPath string, _readFile actionYamlReader, _writeFile fileWriter) (*model.Action, error) {
+ return &model.Action{}, nil
+ }})
+ }
+ return preparers
}
func TestPrintPrepareActionsSkipsWithoutActions(t *testing.T) {
@@ -163,26 +227,6 @@ func TestPrintPrepareActionsSkipsWithoutActions(t *testing.T) {
assert.Empty(t, buf.String())
}
-func TestPrintPrepareActionsFailsJobOnDownloadError(t *testing.T) {
- logger := logging.New(logtest.New())
- ctx := common.WithJobErrorContainer(common.WithLogger(t.Context(), logger.With("job", "j1")))
-
- downloadErr := errors.New("failed to fetch \"actions/checkout\"")
- rc := &RunContext{}
- remaining := &actionPreparerMock{reference: "actions/setup-go@v6", sha: "sha"}
-
- err := printPrepareActions(rc, []actionPreparer{
- &actionPreparerMock{err: downloadErr},
- remaining,
- })(ctx)
-
- require.ErrorIs(t, err, downloadErr)
- // No step has run yet, so the failure has to be recorded against the job itself.
- assert.Equal(t, downloadErr, common.JobError(ctx))
- assert.True(t, rc.jobFailed)
- assert.Zero(t, remaining.prepared)
-}
-
func TestNewStepsExecutorSkipsStepsAfterSetupFailure(t *testing.T) {
for name, tt := range map[string]struct {
preErr error
diff --git a/act/runner/run_context.go b/act/runner/run_context.go
index 0ec7eb3e..6c7c903a 100644
--- a/act/runner/run_context.go
+++ b/act/runner/run_context.go
@@ -82,9 +82,9 @@ type RunContext struct {
platformImage string // container.image or the runs-on pick, resolved once by isEnabled
jobIndex, jobTotal int // strategy.job-index and job-total, total 0 when unknown
- jobEnv map[string]string // workflow and job env as evaluated for the container and service specs
- dockerActionImages map[*model.Step]string // Dockerfile action images built for this job, removed when it ends
- actionDownloads map[string]actionDownload // remote action repositories by clone URL and ref, fetched once per job
+ jobEnv map[string]string // workflow and job env as evaluated for the container and service specs
+ dockerActionImages map[*model.Step]string // Dockerfile action images built for this job, removed when it ends
+ actionDownloads map[string]*actionDownload // remote action repositories by clone URL and ref, fetched once per job
// summaryFileInitialized tracks which per-step summary files (workflow/step-summary-N.md)
// have already been created on the JobContainer. The runner sets up file-command files
@@ -922,7 +922,7 @@ func (rc *RunContext) stopServiceContainers() common.Executor {
const defaultServiceReadyTimeout = 5 * time.Minute
-const serviceReadyPollInterval = time.Second
+const serviceReadyPollInterval = 250 * time.Millisecond
// reportUnstartedServices logs a service that did not start. The steps that need it
// report it better than the runner can, so the job carries on.
diff --git a/act/runner/run_context_test.go b/act/runner/run_context_test.go
index 9e5116ef..3cada89d 100644
--- a/act/runner/run_context_test.go
+++ b/act/runner/run_context_test.go
@@ -273,6 +273,8 @@ func startJobContainerInputs(t *testing.T, workflowYAML string, cfg *Config) []*
w.WriteHeader(http.StatusNoContent)
case strings.HasSuffix(r.URL.Path, "/info"):
_, _ = io.WriteString(w, `{"Architecture":"amd64","OSType":"linux"}`)
+ case strings.Contains(r.URL.Path, "/containers/") && strings.HasSuffix(r.URL.Path, "/json"):
+ http.NotFound(w, r)
default:
t.Errorf("unexpected Docker request: %s %s", r.Method, r.URL)
http.NotFound(w, r)
@@ -1475,7 +1477,7 @@ func TestWaitForServiceContainers(t *testing.T) {
var output bytes.Buffer
logger := logging.New(logging.NewLevelHandler(newJobLogHandler(&output, cyan), logging.LevelInfo))
require.NoError(t, newRunContext(0, &serviceContainer{name: "postgres", container: service}).waitForServiceContainers()(common.WithLogger(t.Context(), logger)))
- assert.Equal(t, []time.Duration{0, time.Second, 2 * time.Second}, pollTimes)
+ assert.Equal(t, []time.Duration{0, serviceReadyPollInterval, 2 * serviceReadyPollInterval}, pollTimes)
assert.Equal(t, 1, strings.Count(output.String(), "postgres service is starting."))
assert.Contains(t, output.String(), "postgres service is healthy.")
})
diff --git a/act/runner/runner.go b/act/runner/runner.go
index ec2cb9bc..b85f4d12 100644
--- a/act/runner/runner.go
+++ b/act/runner/runner.go
@@ -335,7 +335,9 @@ func (runner *runnerImpl) NewPlanExecutor(plan *model.Plan) common.Executor {
}
executor := common.NewPipelineExecutor(stagePipeline...).Then(handleFailure(plan))
- return func(ctx context.Context) error { return executor(ghcontext.WithPinnedCheckout(ctx)) }
+ return func(ctx context.Context) error {
+ return executor(container.WithRunnerArch(ghcontext.WithPinnedCheckout(ctx)))
+ }
}
func handleFailure(plan *model.Plan) common.Executor {
diff --git a/act/runner/step.go b/act/runner/step.go
index 9d84c779..86326983 100644
--- a/act/runner/step.go
+++ b/act/runner/step.go
@@ -165,11 +165,14 @@ func runStepExecutor(step step, stage stepStage, executor common.Executor) commo
if topRC.summaryFileInitialized == nil {
topRC.summaryFileInitialized = map[int]bool{}
}
- _ = rc.JobContainer.Copy(actPath, files...)(ctx)
+ initialFiles := files
if !topRC.summaryFileInitialized[stepSummaryIndex] {
- _ = rc.JobContainer.Copy(actPath, &container.FileEntry{Name: summaryFileCommand, Mode: 0o666})(ctx)
+ initialFiles = append([]*container.FileEntry{{Name: summaryFileCommand, Mode: 0o666}}, files...)
topRC.summaryFileInitialized[stepSummaryIndex] = true
}
+ if err := rc.JobContainer.Copy(actPath, initialFiles...)(ctx); err != nil && len(initialFiles) > len(files) {
+ _ = rc.JobContainer.Copy(actPath, files...)(ctx) // the summary goes first and the resets retry alone, so neither failure blocks the other
+ }
// The command handler needs the step's env to judge ACTIONS_ALLOW_UNSECURE_COMMANDS.
// Cloned: the step executor keeps writing to its own env map after this point, on a
diff --git a/act/runner/step_action_remote.go b/act/runner/step_action_remote.go
index 051dd7e2..dd5f4880 100644
--- a/act/runner/step_action_remote.go
+++ b/act/runner/step_action_remote.go
@@ -31,16 +31,32 @@ type stepActionRemote struct {
env map[string]string
remoteAction *remoteAction
actionDir string // the repository at the resolved commit, shared by the job's actions in it
- resolvedSha string // set on the step that made the job's download of the repository
+ resolvedSha string // set on the first step preparing the repository
}
type actionDownload struct {
- dir string
- err error
+ dir, sha string
+ err error
+ fetched chan struct{}
}
var stepActionRemoteFetch = git.FetchAction
+func (sar *stepActionRemote) fetchAction(ctx context.Context, remote *remoteAction, token string) (string, string, error) {
+ cloneURL := remote.CloneURL(sar.RunContext.Config.DefaultActionURL())
+ return stepActionRemoteFetch(ctx, sar.RunContext.Config.withInstanceAuth(git.NewGitCloneExecutorInput{
+ URL: cloneURL,
+ Ref: remote.Ref,
+ Dir: filepath.Join(sar.RunContext.ActionCacheDir(), model.UsesHash(remote.downloadKey(sar.RunContext.Config.DefaultActionURL()))),
+ OfflineMode: sar.RunContext.Config.ActionOfflineMode,
+ Depth: sar.RunContext.Config.ActionCloneDepth,
+ }, token))
+}
+
+func (sar *stepActionRemote) skipsLocalCheckout(remote *remoteAction, github *model.GithubContext) bool {
+ return remote.IsCheckout() && isLocalCheckout(github, sar.Step) && !sar.RunContext.Config.NoSkipCheckout
+}
+
func (sar *stepActionRemote) prepareActionExecutor() common.Executor {
return func(ctx context.Context) error {
if sar.remoteAction != nil && sar.action != nil {
@@ -63,29 +79,30 @@ func (sar *stepActionRemote) prepareActionExecutor() common.Executor {
return err
}
- if sar.remoteAction.IsCheckout() && isLocalCheckout(github, sar.Step) && !sar.RunContext.Config.NoSkipCheckout {
+ if sar.skipsLocalCheckout(sar.remoteAction, github) {
common.Logger(ctx).Debugf("Skipping local actions/checkout because workdir was already copied")
return nil
}
- cloneURL := sar.remoteAction.CloneURL(sar.RunContext.Config.DefaultActionURL())
job := sar.RunContext.topLevelRunContext()
- key := cloneURL + "@" + sar.remoteAction.Ref
- download, ok := job.actionDownloads[key]
- if !ok {
+ key := sar.remoteAction.downloadKey(sar.RunContext.Config.DefaultActionURL())
+ download := job.actionDownloads[key]
+ if download == nil {
+ download = &actionDownload{}
// For Gitea
// github.Token, unlike Config.GetToken, survives the composite config copy, which nils Config.Secrets.
- download.dir, sar.resolvedSha, download.err = stepActionRemoteFetch(ctx, sar.RunContext.Config.withInstanceAuth(git.NewGitCloneExecutorInput{
- URL: cloneURL,
- Ref: sar.remoteAction.Ref,
- Dir: filepath.Join(sar.RunContext.ActionCacheDir(), model.UsesHash(key)),
- OfflineMode: sar.RunContext.Config.ActionOfflineMode,
- Depth: sar.RunContext.Config.ActionCloneDepth,
- }, github.Token))
+ download.dir, download.sha, download.err = sar.fetchAction(ctx, sar.remoteAction, github.Token)
if job.actionDownloads == nil {
- job.actionDownloads = map[string]actionDownload{}
+ job.actionDownloads = map[string]*actionDownload{}
}
job.actionDownloads[key] = download
+ } else if download.fetched != nil {
+ select {
+ case <-download.fetched:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
}
+ sar.resolvedSha, download.sha = download.sha, ""
var refErr *git.Error
switch {
case errors.As(download.err, &refErr) && errors.Is(download.err, git.ErrShortRef):
@@ -137,7 +154,7 @@ func (sar *stepActionRemote) main() common.Executor {
defer rawLogger.Infof("::endgroup::")
github := sar.getGithubContext(ctx)
- if sar.remoteAction.IsCheckout() && isLocalCheckout(github, sar.Step) && !sar.RunContext.Config.NoSkipCheckout {
+ if sar.skipsLocalCheckout(sar.remoteAction, github) {
if sar.RunContext.Config.BindWorkdir {
common.Logger(ctx).Debugf("Skipping local actions/checkout because you bound your workspace")
return nil
@@ -188,7 +205,7 @@ func (sar *stepActionRemote) getIfExpression(ctx context.Context, stage stepStag
switch stage {
case stepStagePre:
github := sar.getGithubContext(ctx)
- if sar.remoteAction.IsCheckout() && isLocalCheckout(github, sar.Step) && !sar.RunContext.Config.NoSkipCheckout {
+ if sar.skipsLocalCheckout(sar.remoteAction, github) {
// skip local checkout pre step
return "false"
}
@@ -244,6 +261,10 @@ type remoteAction struct {
Ref string
}
+func (ra *remoteAction) downloadKey(defaultURL string) string {
+ return ra.CloneURL(defaultURL) + "@" + ra.Ref
+}
+
func (ra *remoteAction) CloneURL(u string) string {
if ra.URL == "" {
// keep an absolute local path as-is (used by tests to resolve actions from a local
diff --git a/act/runner/step_action_remote_test.go b/act/runner/step_action_remote_test.go
index 1c40368c..9c8fd10b 100644
--- a/act/runner/step_action_remote_test.go
+++ b/act/runner/step_action_remote_test.go
@@ -785,6 +785,7 @@ func TestStepActionRemotePreResolvesDownloadedCommit(t *testing.T) {
readAction: readActionImpl,
}
+ require.NoError(t, printPrepareActions(sar.RunContext, []actionPreparer{sar})(t.Context()))
require.NoError(t, sar.prepareActionExecutor()(t.Context()))
reference, sha := sar.actionDownloadInfo()
@@ -876,7 +877,7 @@ func TestStepActionRemoteCloneTokenSurvivesNilSecrets(t *testing.T) {
sarm.On("readAction", sar.Step, actionDir, "", mock.Anything, mock.Anything).Return(&model.Action{}, nil)
- err := sar.prepareActionExecutor()(ctx)
+ err := printPrepareActions(sar.RunContext, []actionPreparer{sar})(ctx)
require.NoError(t, err)
assert.Equal(t, tt.wantCloneToken, capturedToken)
diff --git a/act/runner/step_test.go b/act/runner/step_test.go
index 7e9e028e..5dee858a 100644
--- a/act/runner/step_test.go
+++ b/act/runner/step_test.go
@@ -9,6 +9,7 @@ import (
"errors"
"io"
"os"
+ "path/filepath"
"strings"
"testing"
"testing/synctest"
@@ -540,10 +541,11 @@ func TestRunStepExecutorParity(t *testing.T) {
}
})
- t.Run("file command error honors continue-on-error", func(t *testing.T) {
+ t.Run("file command error honors continue-on-error and keeps the step summary", func(t *testing.T) {
step := newStep(t, &model.Step{ID: "commands", RawContinueOnError: "true"})
logger, hook := logtest.NewLogger()
err := runStepExecutor(step, stepStageMain, func(context.Context) error {
+ require.NoError(t, os.WriteFile(step.env["GITHUB_STEP_SUMMARY"], []byte("report"), 0o600))
require.NoError(t, os.WriteFile(step.env["GITHUB_ENV"], []byte("GOOD=1\nmalformed\n"), 0o600))
require.NoError(t, os.WriteFile(step.env["GITHUB_OUTPUT"], []byte("kept=value\n"), 0o600))
require.NoError(t, os.WriteFile(step.env["GITHUB_STATE"], []byte("saved=value\n"), 0o600))
@@ -563,6 +565,15 @@ func TestRunStepExecutorParity(t *testing.T) {
require.NoError(t, readErr)
assert.Empty(t, contents)
}
+ summary, err := os.ReadFile(step.env["GITHUB_STEP_SUMMARY"])
+ require.NoError(t, err)
+ assert.Equal(t, "report", string(summary))
+ })
+
+ t.Run("a broken summary path still resets the command files", func(t *testing.T) {
+ step := newStep(t, &model.Step{ID: "noop"})
+ require.NoError(t, os.MkdirAll(filepath.Join(step.RunContext.JobContainer.GetActPath(), "workflow", "step-summary-0.md"), 0o700))
+ require.NoError(t, runStepExecutor(step, stepStageMain, noopExecutor)(t.Context()))
})
t.Run("file commands survive a job timeout", func(t *testing.T) {
diff --git a/internal/app/run/runner_test.go b/internal/app/run/runner_test.go
index 1ab34541..0e1cf4e9 100644
--- a/internal/app/run/runner_test.go
+++ b/internal/app/run/runner_test.go
@@ -156,6 +156,8 @@ func TestRunnerReclaimsVolumesAfterReporting(t *testing.T) {
removed.Add(1)
}
writer.WriteHeader(http.StatusNoContent)
+ case strings.HasPrefix(path, "/containers/") && strings.HasSuffix(path, "/json"):
+ http.NotFound(writer, request)
default:
t.Errorf("unexpected Docker request: %s %s", request.Method, request.URL)
http.NotFound(writer, request)