From 1a7f09224f3f195b7e39766feb674b300c5b154a Mon Sep 17 00:00:00 2001 From: silverwind <2021+silverwind@noreply.gitea.com> Date: Wed, 21 Oct 2026 05:12:54 +0000 Subject: [PATCH] perf: speed up job setup, action preparation and the cache server (#1327) - Look up containers by name instead of listing all of them - On Linux, cache the docker proxy discovery instead of starting a probe container per job - Download a job's actions concurrently and share identical fetches between concurrent jobs - Cache `runner.arch` per run and poll service health every 250ms - Cache server: queue on an in-process lock instead of polling the database file lock, upload S3 objects up to 64MiB in one request and larger ones in 4 concurrent parts | `exec`, median of 5 | Before | After | |---|---:|---:| | 1 step | 1.70s | 0.87s | | 10 steps | 2.84s | 1.97s | | Service + step | 3.78s | 2.18s | | checkout + Node action | 3.64s | 2.52s | Reviewed-on: https://gitea.com/gitea/runner/pulls/1327 Reviewed-by: Zettat123 <39446+zettat123@noreply.gitea.com> --- act/artifactcache/handler.go | 74 +++++++---- act/artifactcache/handler_test.go | 22 ++++ act/artifactcache/s3_storage.go | 42 +++++-- act/artifactcache/s3_storage_test.go | 59 ++++++++- act/common/executor.go | 14 ++- act/common/git/fetch_action.go | 41 ++++++ act/common/git/fetch_action_test.go | 90 +++++++++++++ act/container/docker_images.go | 12 +- act/container/docker_images_test.go | 7 ++ act/container/docker_proxy.go | 30 ++++- act/container/docker_proxy_test.go | 43 +++++++ act/container/docker_run.go | 23 ++-- act/container/docker_run_test.go | 65 ++++++++-- .../linux_container_environment_extensions.go | 33 ++++- act/runner/expression.go | 2 + act/runner/expression_test.go | 20 +++ act/runner/job_executor.go | 35 ++++++ act/runner/job_executor_test.go | 118 ++++++++++++------ act/runner/run_context.go | 8 +- act/runner/run_context_test.go | 4 +- act/runner/runner.go | 4 +- act/runner/step.go | 7 +- act/runner/step_action_remote.go | 57 ++++++--- act/runner/step_action_remote_test.go | 3 +- act/runner/step_test.go | 13 +- internal/app/run/runner_test.go | 2 + 26 files changed, 688 insertions(+), 140 deletions(-) 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)