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 <[email protected]>

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

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()

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 {

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("<InitiateMultipartUploadResult><UploadId>upload</UploadId></InitiateMultipartUploadResult>"))
+			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("<Error><Code>InvalidPart</Code></Error>"))
+				}
+			case req.Method == http.MethodPost:
+				require.NoError(t, xml.NewDecoder(req.Body).Decode(&completion))
+				response.Body = io.NopCloser(strings.NewReader("<CompleteMultipartUploadResult/>"))
+			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)