fix(cache): stop the cache server from running its volume out of space (#1377)

Committing a cache upload copied every part into a new archive before deleting the parts, so a 5 GB save needed 10 GB free. Failed writes and assemblies left their partial bytes until the next stale sweep, up to an hour later, and every other save on the runner failed meanwhile. Nothing made room for an upload, so once the cache had filled its volume down to the free space floor, larger saves kept failing.

- Assemble in place, deleting each part once appended.
- Drop partial parts and archives on failure.
- Before a part would cross the floor, evict the least recently used entries.
- Reject uploads to reservations the sweep already deleted, which left files nothing removed.

_Written by Claude (Opus 5.5)._

Reviewed-on: https://gitea.com/gitea/runner/pulls/1377
Reviewed-by: bircni <[email protected]>
Co-authored-by: silverwind <[email protected]>
This commit is contained in:
silverwind authored and bircni committed 2026-11-04 16:38:21 +00:00
1 parent dcba546cb3
commit 3e1b4b42c2
6 files changed
+60 -51

No files matched your search

+14 -13
View File
@@ -124,7 +124,7 @@ type Handler struct {
policy Policy policy Policy
// freeDisk is a field so tests can drive evictForFreeSpace without a full volume. // freeDisk is a field so tests can drive makeRoom without a full volume.
freeDisk func(string) (uint64, error) freeDisk func(string) (uint64, error)
} }
@@ -773,6 +773,7 @@ func (h *Handler) upload(w http.ResponseWriter, r *http.Request) {
h.responseJSON(w, r, 400, err) h.responseJSON(w, r, 400, err)
return return
} }
h.makeRoom(r.ContentLength)
if err := h.storage.Write(cache.ID, start, r.Body); err != nil { if err := h.storage.Write(cache.ID, start, r.Body); err != nil {
h.responseJSON(w, r, 500, err) h.responseJSON(w, r, 500, err)
return return
@@ -875,11 +876,8 @@ func (h *Handler) commitCache(cache *Cache) error {
if h.s3 != nil { if h.s3 != nil {
return nil // lifecycle rules on the bucket bound its size 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) h.evictRepo(db, cache.Repo)
h.evictTotal(db) h.evictTotal(db)
h.evictForFreeSpace(db)
return nil return nil
} }
@@ -1173,9 +1171,6 @@ func (h *Handler) touchCache(id uint64, requireIncomplete bool) error {
defer db.Close() defer db.Close()
cache, err := db.get(id) cache, err := db.get(id)
if errors.Is(err, errCacheNotFound) {
return nil
}
if err != nil { if err != nil {
return err return err
} }
@@ -1260,23 +1255,29 @@ func (h *Handler) gcCache() {
h.evictExpired(db) h.evictExpired(db)
h.evictSuperseded(db) h.evictSuperseded(db)
h.evictOversized(db) h.evictOversized(db)
h.evictForFreeSpace(db) _ = db.Close()
h.makeRoom(0)
} }
// evictForFreeSpace bounds the volume itself, so it also covers bytes the cache never // makeRoom evicts the least recently used entries to keep need bytes above the floor, as GitHub saves new over old.
// accounted for. func (h *Handler) makeRoom(need int64) {
func (h *Handler) evictForFreeSpace(db *cacheDB) {
free, err := h.freeDisk(h.dir) free, err := h.freeDisk(h.dir)
if err != nil { if err != nil {
h.logger.Debugf("free disk check: %v", err) // unsupported platform, treat as unavailable rather than full h.logger.Debugf("free disk check: %v", err) // unsupported platform, treat as unavailable rather than full
return return
} }
if free >= uint64(h.policy.MinFreeDisk) { shortfall := h.policy.MinFreeDisk + max(need, 0) - int64(free)
if h.s3 != nil || shortfall <= 0 {
return return
} }
db, err := h.openDB()
if err != nil {
return
}
defer db.Close()
caches := h.completedByUse(db) caches := h.completedByUse(db)
total, shortfall := totalSize(caches), h.policy.MinFreeDisk-int64(free) total := totalSize(caches)
if total <= shortfall { if total <= shortfall {
// Say so, or shedding everything and still being short reads as the backstop working. // Say so, or shedding everything and still being short reads as the backstop working.
h.logger.Warnf("cache volume is %d MiB short of the free space floor with only %d MiB of cache on it; something else is filling it", shortfall/miB, total/miB) h.logger.Warnf("cache volume is %d MiB short of the free space floor with only %d MiB of cache on it; something else is filling it", shortfall/miB, total/miB)
+15 -9
View File
@@ -18,6 +18,7 @@ import (
"strconv" "strconv"
"strings" "strings"
"testing" "testing"
"testing/iotest"
"testing/synctest" "testing/synctest"
"time" "time"
@@ -673,7 +674,7 @@ func TestHandler_evictPolicy(t *testing.T) {
} }
} }
func TestHandler_evictForFreeSpace(t *testing.T) { func TestHandler_makeRoom(t *testing.T) {
t.Parallel() t.Parallel()
free := func(n int64) func(string) (uint64, error) { free := func(n int64) func(string) (uint64, error) {
return func(string) (uint64, error) { return uint64(n), nil } return func(string) (uint64, error) { return uint64(n), nil }
@@ -682,14 +683,16 @@ func TestHandler_evictForFreeSpace(t *testing.T) {
for _, tc := range []struct { for _, tc := range []struct {
name string name string
freeDisk func(string) (uint64, error) freeDisk func(string) (uint64, error)
upload int64
kept []string kept []string
}{ }{
{"ample free space evicts nothing", free(defaultMinFreeDisk), []string{"oldest", "middle", "newest"}}, {"ample free space evicts nothing", free(defaultMinFreeDisk), 0, []string{"oldest", "middle", "newest"}},
{"a small shortfall sheds one entry", free(defaultMinFreeDisk - 4*miB), []string{"middle", "newest"}}, {"room for an upload sheds the oldest entry", free(defaultMinFreeDisk), 4 * miB, []string{"middle", "newest"}},
{"a shortfall the cache cannot cover sheds all of it", free(0), nil}, {"a shortfall the cache cannot cover sheds all of it", free(0), 0, nil},
{ {
"an unreadable volume is treated as unavailable, not as full", "an unreadable volume is treated as unavailable, not as full",
func(string) (uint64, error) { return 0, errors.New("unsupported") }, func(string) (uint64, error) { return 0, errors.New("unsupported") },
0,
[]string{"oldest", "middle", "newest"}, []string{"oldest", "middle", "newest"},
}, },
} { } {
@@ -702,11 +705,7 @@ func TestHandler_evictForFreeSpace(t *testing.T) {
} }
handler := newTestHandler(t, Policy{}, entries...) handler := newTestHandler(t, Policy{}, entries...)
handler.freeDisk = tc.freeDisk handler.freeDisk = tc.freeDisk
handler.makeRoom(tc.upload)
db, err := handler.openDB()
require.NoError(t, err)
handler.evictForFreeSpace(db)
require.NoError(t, db.Close())
assert.ElementsMatch(t, tc.kept, keptKeys(t, handler, entries)) assert.ElementsMatch(t, tc.kept, keptKeys(t, handler, entries))
}) })
@@ -888,6 +887,13 @@ func keptKeys(t *testing.T, handler *Handler, entries []*Cache) []string {
return kept return kept
} }
func TestStorageDropsFailedWrites(t *testing.T) {
storage, err := NewStorage(t.TempDir())
require.NoError(t, err)
require.Error(t, storage.Write(1, 0, io.MultiReader(strings.NewReader("partial"), iotest.ErrReader(io.ErrUnexpectedEOF))))
assert.NoFileExists(t, storage.tempName(1, 0))
}
func writeBlob(t *testing.T, handler *Handler, id uint64) { func writeBlob(t *testing.T, handler *Handler, id uint64) {
t.Helper() t.Helper()
require.NoError(t, handler.storage.Write(id, 0, strings.NewReader("a"))) require.NoError(t, handler.storage.Write(id, 0, strings.NewReader("a")))
+1
View File
@@ -218,6 +218,7 @@ func (h *Handler) v2UploadBlob(w http.ResponseWriter, r *http.Request) {
h.responseJSON(w, r, http.StatusBadRequest, err) h.responseJSON(w, r, http.StatusBadRequest, err)
return return
} }
h.makeRoom(r.ContentLength)
query := r.URL.Query() query := r.URL.Query()
switch strings.ToLower(query.Get("comp")) { switch strings.ToLower(query.Get("comp")) {
+3
View File
@@ -14,6 +14,7 @@ import (
"strconv" "strconv"
"strings" "strings"
"testing" "testing"
"time"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
@@ -84,6 +85,8 @@ func TestCacheServiceV2RoundTrip(t *testing.T) {
unsigned := fmt.Sprintf("%s%s/1", handler.ExternalURL(), blobPath) unsigned := fmt.Sprintf("%s%s/1", handler.ExternalURL(), blobPath)
assert.Equal(t, http.StatusUnauthorized, putBlob(t, unsigned, content)) assert.Equal(t, http.StatusUnauthorized, putBlob(t, unsigned, content))
unreserved := handler.signedURL(JobCredential{Repo: testRepo}, blobPath, blobUploadPurpose, 1, time.Now().Add(time.Minute))
assert.Equal(t, http.StatusBadRequest, putBlob(t, unreserved, content))
finalized, uploadURL := saveV2(t, handler, "deps-v1", "abc123", content) finalized, uploadURL := saveV2(t, handler, "deps-v1", "abc123", content)
require.Equal(t, true, finalized["ok"]) require.Equal(t, true, finalized["ok"])
+25 -27
View File
@@ -9,6 +9,7 @@ import (
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json/v2" "encoding/json/v2"
"errors"
"fmt" "fmt"
"io" "io"
"net/http" "net/http"
@@ -51,9 +52,10 @@ func (s *Storage) writeFile(name string, reader io.Reader) error {
if err != nil { if err != nil {
return err return err
} }
defer file.Close()
_, err = io.Copy(file, reader) _, err = io.Copy(file, reader)
if err = errors.Join(err, file.Close()); err != nil {
_ = os.Remove(name)
}
return err return err
} }
@@ -102,50 +104,46 @@ func (s *Storage) Commit(id uint64, size int64) (int64, error) {
return 0, err return 0, err
} }
written, err := assemble(name, tempNames) written, err := assemble(name, tempNames)
if err != nil { if err == nil && size >= 0 && written != size { // old clients like actions/cache@v2 send no size
return 0, err err = fmt.Errorf("broken file: %v != %v", written, size)
} }
// If size is less than 0, it means the size is unknown. if err != nil {
// We can't check the size of the file, just skip the check.
// It happens when the request comes from old versions of actions, like `actions/cache@v2`.
if size >= 0 && written != size {
_ = os.Remove(name) _ = os.Remove(name)
return 0, fmt.Errorf("broken file: %v != %v", written, size) return 0, err
} }
return written, nil return written, nil
} }
// assemble concatenates the uploaded parts into name. A single part, which is what the v2 API // assemble appends the parts into the first, deleting each once copied, so it never needs twice the space.
// produces below the client's block threshold, is already the whole archive and is moved.
func assemble(name string, tempNames []string) (int64, error) { func assemble(name string, tempNames []string) (int64, error) {
if len(tempNames) == 1 { if len(tempNames) == 0 {
info, err := os.Stat(tempNames[0]) return 0, errors.New("nothing was uploaded")
if err != nil {
return 0, err
}
return info.Size(), os.Rename(tempNames[0], name)
} }
if err := os.Rename(tempNames[0], name); err != nil {
file, err := os.Create(name) return 0, err
}
file, err := os.OpenFile(name, os.O_WRONLY|os.O_APPEND, 0)
if err != nil { if err != nil {
return 0, err return 0, err
} }
defer file.Close() defer file.Close()
for _, v := range tempNames[1:] {
var written int64 part, err := os.Open(v)
for _, v := range tempNames {
f, err := os.Open(v)
if err != nil { if err != nil {
return 0, err return 0, err
} }
n, err := io.Copy(file, f) _, err = io.Copy(file, part)
_ = f.Close() _ = part.Close()
if err != nil { if err != nil {
return 0, err return 0, err
} }
written += n _ = os.Remove(v)
} }
return written, nil info, err := file.Stat()
if err != nil {
return 0, err
}
return info.Size(), file.Close()
} }
func (s *Storage) Serve(w http.ResponseWriter, r *http.Request, id uint64) { func (s *Storage) Serve(w http.ResponseWriter, r *http.Request, id uint64) {
+2 -2
View File
@@ -206,8 +206,8 @@ cache:
# How the cache server discards entries, ignored when external_server or s3 is set since # How the cache server discards entries, ignored when external_server or s3 is set since
# those apply their own. Leave a setting out for its default; 0s or 0 turns the three # those apply their own. Leave a setting out for its default; 0s or 0 turns the three
# limits off. Sizes accept 10GB, 512mb, 1TiB or a plain byte count, binary either way. # limits off. Sizes accept 10GB, 512mb, 1TiB or a plain byte count, binary either way.
# Whatever these allow, the cache still sheds entries to keep free space on its volume # Whatever these allow, the cache still sheds entries to keep free space on its volume, room
# above health_check.min_free_disk_space_mb when health checks are enabled. # for uploads included, above health_check.min_free_disk_space_mb when health checks are enabled.
# Remove entries nothing has read or written within this window. Only last access counts. # Remove entries nothing has read or written within this window. Only last access counts.
#retention: 168h #retention: 168h
# Cap one repository, removing its least recently accessed entries until it fits. An entry # Cap one repository, removing its least recently accessed entries until it fits. An entry