feat(cache): add shared S3-compatible storage (#1244)

Store the cache used by `actions/cache` in an S3-compatible bucket such as AWS S3 or RustFS, with no `cache-server`. Runners with the same settings use the same cache. A cache saved on one runner can be restored on the others within 30 seconds. The runner talks to the bucket over the S3 REST API with SigV4, without an SDK. Lifecycle rules on the bucket expire old caches. *Description rewritten by Claude.* --------- Co-authored-by: silverwind <[email protected]> Reviewed-on: https://gitea.com/gitea/runner/pulls/1244 Reviewed-by: silverwind <[email protected]> Co-authored-by: bircni <[email protected]>
2026-11-07 07:40:29 +00:00 · 2026-09-24 19:46:12 +00:00
1 parent ace8653105
commit 81add27459
@@ -10,6 +10,7 @@ import (
 	"crypto/hmac"
 	"crypto/rand"
 	"crypto/sha256"
+	"encoding/binary"
 	"encoding/hex"
 	"encoding/json/v2"
 	"errors"
@@ -78,7 +79,8 @@ type credEntry struct {

 type Handler struct {
 	dir      string
-	storage  *Storage
+	storage  cacheStorage
+	s3       *s3Storage
 	router   *httprouter.Router
 	listener net.Listener
 	port     int
@@ -88,6 +90,9 @@ type Handler struct {
 	gcing atomic.Bool
 	gcAt  time.Time

+	stopSync context.CancelFunc
+	syncing  sync.WaitGroup
+
 	outboundIP string

 	// internalSecret guards /_internal/{register,revoke}. When set, a remote
@@ -117,6 +122,7 @@ type Options struct {
 	Dir        string
 	OutboundIP string
 	Port       uint16
+	S3         *S3Options
 	Upstream   string

 	// InternalSecret, when non-empty, enables a control-plane API at
@@ -185,6 +191,12 @@ func StartHandler(opts Options) (*Handler, error) {
 		return nil, err
 	}
 	h.storage = storage
+	if opts.S3 != nil {
+		if h.s3, err = newS3Storage(storage, *opts.S3); err != nil {
+			return nil, err
+		}
+		h.storage = h.s3
+	}

 	secret, err := loadOrCreateSecret(dir)
 	if err != nil {
@@ -216,6 +228,11 @@ func StartHandler(opts Options) (*Handler, error) {
 	if err := h.serve(opts.Port, router); err != nil {
 		return nil, err
 	}
+	if h.s3 != nil {
+		ctx, cancel := context.WithCancel(context.Background())
+		h.stopSync = cancel
+		h.syncing.Go(func() { h.syncMetadata(ctx) })
+	}
 	return h, nil
 }

@@ -380,6 +397,10 @@ func (h *Handler) Close() error {
 	if h == nil {
 		return nil
 	}
+	if h.stopSync != nil {
+		h.stopSync()
+		h.syncing.Wait()
+	}
 	var retErr error
 	if h.server != nil {
 		err := h.server.Close()
@@ -413,6 +434,72 @@ func (h *Handler) openDB() (*bolthold.Store, error) {
 	})
 }

+// syncMetadata mirrors the bucket into the store until ctx ends, so no request waits on S3 for it.
+func (h *Handler) syncMetadata(ctx context.Context) {
+	ticker := time.NewTicker(metadataSyncInterval)
+	defer ticker.Stop()
+	for {
+		if err := h.importMetadata(ctx); err != nil && ctx.Err() == nil {
+			h.logger.Warnf("sync S3 cache: %v", err)
+		}
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
+// importMetadata lists after snapshotting the store, so an entry committed meanwhile, published before it is stored, stays.
+func (h *Handler) importMetadata(ctx context.Context) error {
+	db, err := h.openDB()
+	if err != nil {
+		return err
+	}
+	local := h.caches(db, bolthold.Where("Complete").Eq(true).Index("Complete"))
+	db.Close()
+
+	published, err := h.s3.published(ctx)
+	if err != nil {
+		return err
+	}
+	var gone []*Cache
+	for _, cache := range local {
+		if !published[cache.ID] {
+			gone = append(gone, cache)
+		}
+		delete(published, cache.ID)
+	}
+	added := make([]*Cache, 0, len(published))
+	for id := range published {
+		cache, err := h.s3.fetch(ctx, id)
+		if err != nil {
+			return err
+		}
+		added = append(added, cache)
+	}
+
+	db, err = h.openDB()
+	if err != nil {
+		return err
+	}
+	defer db.Close()
+	return db.Bolt().Update(func(tx *bbolt.Tx) error {
+		for _, cache := range added {
+			if err := db.TxUpsert(tx, cache.ID, cache); err != nil {
+				return err
+			}
+		}
+		for _, cache := range gone {
+			_ = h.s3.Storage.Remove(cache.ID) // a blob kept on disk before the runner used S3
+			if err := db.TxDelete(tx, cache.ID, cache); err != nil && !errors.Is(err, bolthold.ErrNotFound) {
+				return err
+			}
+		}
+		return nil
+	})
+}
+
 // GET /_apis/artifactcache/cache
 func (h *Handler) find(w http.ResponseWriter, r *http.Request, _ httprouter.Params) {
 	cred := credFromContext(r.Context())
@@ -455,7 +542,7 @@ func (h *Handler) lookupCache(db *bolthold.Store, repo string, keys []string, ve
 		return nil, err
 	}
 	if !ok {
-		_ = db.Delete(cache.ID, cache)
+		h.deleteCache(db, cache)
 		return nil, nil //nolint:nilnil // absence is not an error here
 	}
 	// Handing out a download URL counts as access, or eviction could drop the entry between
@@ -487,7 +574,7 @@ func (h *Handler) reserve(w http.ResponseWriter, r *http.Request, _ httprouter.P
 	now := time.Now().Unix()
 	cache.CreatedAt = now
 	cache.UsedAt = now
-	if err := insertCache(db, cache); err != nil {
+	if err := h.insertCache(db, cache); err != nil {
 		h.responseJSON(w, r, 500, err)
 		return
 	}
@@ -601,6 +688,12 @@ func (h *Handler) commitCache(cache *Cache) error {
 	cache.Size = written
 	cache.Complete = true
 	cache.UsedAt = time.Now().Unix() // a just-written entry counts as accessed, so it cannot be its own eviction victim
+	if h.s3 != nil {
+		if err := h.s3.publish(cache); err != nil {
+			_ = h.storage.Remove(cache.ID)
+			return err
+		}
+	}

 	db, err := h.openDB()
 	if err != nil {
@@ -610,6 +703,9 @@ func (h *Handler) commitCache(cache *Cache) error {
 	if err := db.Update(cache.ID, cache); err != nil {
 		return err
 	}
+	if h.s3 != nil {
+		return nil // lifecycle rules on the bucket bound its size
+	}
 	// A commit is the only thing that grows the store, so the only thing that can push the
 	// volume under the floor.
 	h.evictRepo(db, cache.Repo)
@@ -873,7 +969,13 @@ func findExactCache(db *bolthold.Store, repo, key, version string, complete bool
 	return cache, nil
 }

-func insertCache(db *bolthold.Store, cache *Cache) error {
+func (h *Handler) insertCache(db *bolthold.Store, cache *Cache) error {
+	if h.s3 != nil {
+		var id [8]byte
+		_, _ = rand.Read(id[:])
+		cache.ID = binary.BigEndian.Uint64(id[:]) >> 11 // IDs must be unique across runners, and v1 clients parse an ID as a JS number
+		return db.Insert(cache.ID, cache)
+	}
 	return db.Bolt().Update(func(tx *bbolt.Tx) error {
 		if err := db.TxInsert(tx, bolthold.NextSequence(), cache); err != nil {
 			return fmt.Errorf("insert cache: %w", err)
@@ -931,6 +1033,8 @@ const (
 	// a stale reservation.
 	uploadStallTimeout = 5 * time.Minute

+	metadataSyncInterval = 30 * time.Second
+
 	defaultMinFreeDisk = 1024 * miB
 )

@@ -978,6 +1082,9 @@ func (h *Handler) gcCache() {
 	defer db.Close()

 	h.evictIncomplete(db)
+	if h.s3 != nil {
+		return // lifecycle rules on the bucket expire its entries
+	}
 	h.evictExpired(db)
 	h.evictSuperseded(db)
 	h.evictOversized(db)
@@ -1175,7 +1282,11 @@ 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 {
-	if err := h.storage.Remove(cache.ID); err != nil {
+	storage := h.storage
+	if h.s3 != nil && !cache.Complete {
+		storage = h.s3.Storage // an upload not yet committed has nothing in the bucket
+	}
+	if err := storage.Remove(cache.ID); err != nil {
 		h.logger.Warnf("remove cache blob: %v", err)
 		return false
 	}