fix: serialize artifact cache upload and commit for the same entry

pkg/artifactcache/cachelock.go

@@ -0,0 +1,53 @@
+package artifactcache
+
+import "sync"
+
+// cacheLock hands out one mutex per cache ID, so callers operating on
+// different cache entries never block each other while callers operating on
+// the same entry are fully serialized.
+//
+// This exists to close the race between the upload and commit handlers
+// described in https://github.com/nektos/act/issues/6012: both close their
+// database handle before touching storage, so nothing previously prevented
+// a commit from finalizing (and, on cleanup, deleting) a cache's temporary
+// chunk directory while a concurrent upload was still writing into it.
+type cacheLock struct {
+	mu    sync.Mutex
+	locks map[uint64]*refCountedMutex
+}
+
+type refCountedMutex struct {
+	mu  sync.Mutex
+	ref int
+}
+
+func newCacheLock() *cacheLock {
+	return &cacheLock{locks: make(map[uint64]*refCountedMutex)}
+}
+
+// Lock blocks until the lock for id is held and returns a function that
+// releases it. The caller must invoke the returned function exactly once,
+// typically via defer, to avoid leaking the entry.
+func (c *cacheLock) Lock(id uint64) func() {
+	c.mu.Lock()
+	l, ok := c.locks[id]
+	if !ok {
+		l = &refCountedMutex{}
+		c.locks[id] = l
+	}
+	l.ref++
+	c.mu.Unlock()
+
+	l.mu.Lock()
+
+	return func() {
+		l.mu.Unlock()
+
+		c.mu.Lock()
+		l.ref--
+		if l.ref == 0 {
+			delete(c.locks, id)
+		}
+		c.mu.Unlock()
+	}
+}

pkg/artifactcache/cachelock_test.go

@@ -0,0 +1,107 @@
+package artifactcache
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCacheLock_SameIDIsSerialized(t *testing.T) {
+	c := newCacheLock()
+
+	unlock := c.Lock(1)
+
+	acquired := make(chan struct{})
+	go func() {
+		unlock := c.Lock(1)
+		defer unlock()
+		close(acquired)
+	}()
+
+	select {
+	case <-acquired:
+		t.Fatal("second Lock(1) returned before the first was released")
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	unlock()
+
+	select {
+	case <-acquired:
+	case <-time.After(time.Second):
+		t.Fatal("second Lock(1) did not return after the first was released")
+	}
+}
+
+func TestCacheLock_DifferentIDsDoNotBlock(t *testing.T) {
+	c := newCacheLock()
+
+	unlock1 := c.Lock(1)
+	defer unlock1()
+
+	done := make(chan struct{})
+	go func() {
+		unlock2 := c.Lock(2)
+		defer unlock2()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("Lock(2) blocked on an unrelated Lock(1)")
+	}
+}
+
+func TestCacheLock_ReleasesMapEntryWhenUncontended(t *testing.T) {
+	c := newCacheLock()
+
+	for i := 0; i < 100; i++ {
+		unlock := c.Lock(42)
+		unlock()
+	}
+
+	c.mu.Lock()
+	n := len(c.locks)
+	c.mu.Unlock()
+	assert.Equal(t, 0, n, "cacheLock leaked map entries for a single uncontended ID")
+}
+
+func TestCacheLock_ConcurrentUseAcrossManyIDs(t *testing.T) {
+	c := newCacheLock()
+
+	var wg sync.WaitGroup
+	const goroutines = 50
+	const idsPerGoroutine = 20
+	for g := 0; g < goroutines; g++ {
+		wg.Add(1)
+		go func(g int) {
+			defer wg.Done()
+			for i := 0; i < idsPerGoroutine; i++ {
+				id := uint64(g%5) * 1000 // deliberately overlap across goroutines
+				unlock := c.Lock(id)
+				unlock()
+			}
+		}(g)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("concurrent Lock/unlock across overlapping IDs did not complete, possible deadlock")
+	}
+
+	c.mu.Lock()
+	n := len(c.locks)
+	c.mu.Unlock()
+	require.Equal(t, 0, n, "cacheLock leaked map entries after all locks were released")
+}

pkg/artifactcache/handler.go

@@ -40,13 +40,17 @@ type Handler struct {
 	gcing atomic.Bool
 	gcAt  time.Time
 
+	cacheLocks *cacheLock
+
 	outboundIP        string
 	customExternalURL string
 	token             string
 }
 
 func StartHandler(dir, customExternalURL string, outboundIP string, port uint16, logger logrus.FieldLogger) (*Handler, error) {
-	h := &Handler{}
+	h := &Handler{
+		cacheLocks: newCacheLock(),
+	}
 
 	if logger == nil {
 		discard := logrus.New()
@@ -245,6 +249,13 @@ func (h *Handler) reserve(w http.ResponseWriter, r *http.Request, _ httprouter.P
 	})
 }
 
+// testHookUploadLocked is called by upload once it holds the per-cache lock
+// and has confirmed the cache is not yet complete, immediately before it
+// writes to storage. It is a no-op outside of tests; test code overrides it
+// to deterministically interleave a concurrent commit at this exact point,
+// exercising the race described in issue #6012 without relying on timing.
+var testHookUploadLocked = func() {}
+
 // PATCH /_apis/artifactcache/caches/:id
 func (h *Handler) upload(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
 	id, err := strconv.ParseUint(params.ByName("id"), 10, 64)
@@ -253,6 +264,13 @@ func (h *Handler) upload(w http.ResponseWriter, r *http.Request, params httprout
 		return
 	}
 
+	// Serialized with commit for the same cache ID: both handlers close their
+	// database handle before touching storage, so without this lock a commit
+	// could finalize (and on cleanup, delete) this cache's temporary chunk
+	// directory while this write is still in flight. See issue #6012.
+	unlock := h.cacheLocks.Lock(id)
+	defer unlock()
+
 	cache := &Cache{}
 	db, err := h.openDB()
 	if err != nil {
@@ -274,6 +292,7 @@ func (h *Handler) upload(w http.ResponseWriter, r *http.Request, params httprout
 		return
 	}
 	db.Close()
+	testHookUploadLocked()
 	start, _, err := parseContentRange(r.Header.Get("Content-Range"))
 	if err != nil {
 		h.responseJSON(w, r, 400, err)
@@ -294,6 +313,10 @@ func (h *Handler) commit(w http.ResponseWriter, r *http.Request, params httprout
 		return
 	}
 
+	// See the matching comment in upload.
+	unlock := h.cacheLocks.Lock(uint64(id))
+	defer unlock()
+
 	cache := &Cache{}
 	db, err := h.openDB()
 	if err != nil {

pkg/artifactcache/handler_test.go

@@ -759,3 +759,113 @@ func TestHandler_BindAddress(t *testing.T) {
 	addr := handler.listener.Addr().String()
 	assert.True(t, strings.HasPrefix(addr, "127.0.0.1:"))
 }
+
+// TestHandler_UploadCommitRace is a regression test for
+// https://github.com/nektos/act/issues/6012: upload and commit for the same
+// cache entry both close their database handle before touching storage, with
+// nothing serializing the two. A commit could finalize the cache -- and, on
+// cleanup, delete its temporary chunk directory -- while a concurrent upload
+// was still writing into it, corrupting or silently dropping the uploaded
+// bytes.
+//
+// The window between upload's completeness check and its storage write is a
+// handful of nanoseconds, far too narrow to hit reliably by chance. Instead
+// this uses testHookUploadLocked to force the exact interleaving every time:
+// it pauses upload right after it has verified the cache is not yet complete
+// and is holding the cache's lock, starts a concurrent commit, confirms
+// commit cannot finish while upload still holds the lock, then releases
+// upload and checks the committed artifact is exactly what was uploaded.
+func TestHandler_UploadCommitRace(t *testing.T) {
+	dir := filepath.Join(t.TempDir(), "artifactcache")
+	handler, err := StartHandler(dir, "", "", 0, nil)
+	require.NoError(t, err)
+	defer handler.Close()
+
+	base := fmt.Sprintf("%s%s", handler.ExternalURL(), apiPath)
+
+	key := strings.ToLower(t.Name())
+	version := "c19da02a2bd7e77277f1ac29ab45c09b7d46a4ee758284e26bb3045ad11d9d20"
+	content := bytes.Repeat([]byte{0x42}, 8192)
+
+	var id uint64
+	{
+		body, err := json.Marshal(&Request{Key: key, Version: version, Size: int64(len(content))})
+		require.NoError(t, err)
+		resp, err := http.Post(fmt.Sprintf("%s/caches", base), "application/json", bytes.NewReader(body))
+		require.NoError(t, err)
+		got := struct {
+			CacheID uint64 `json:"cacheId"`
+		}{}
+		require.NoError(t, json.NewDecoder(resp.Body).Decode(&got))
+		resp.Body.Close()
+		id = got.CacheID
+		require.NotZero(t, id)
+	}
+
+	uploadLocked := make(chan struct{})
+	releaseUpload := make(chan struct{})
+	original := testHookUploadLocked
+	testHookUploadLocked = func() {
+		close(uploadLocked)
+		<-releaseUpload
+	}
+	defer func() { testHookUploadLocked = original }()
+
+	uploadDone := make(chan *http.Response, 1)
+	go func() {
+		req, err := http.NewRequest(http.MethodPatch,
+			fmt.Sprintf("%s/caches/%d", base, id), bytes.NewReader(content))
+		if !assert.NoError(t, err) {
+			uploadDone <- nil
+			return
+		}
+		req.Header.Set("Content-Type", "application/octet-stream")
+		req.Header.Set("Content-Range", fmt.Sprintf("bytes 0-%d/*", len(content)-1))
+		resp, err := http.DefaultClient.Do(req)
+		assert.NoError(t, err)
+		uploadDone <- resp
+	}()
+
+	select {
+	case <-uploadLocked:
+	case <-time.After(5 * time.Second):
+		t.Fatal("upload never reached testHookUploadLocked")
+	}
+
+	commitDone := make(chan *http.Response, 1)
+	go func() {
+		resp, err := http.Post(fmt.Sprintf("%s/caches/%d", base, id), "", nil)
+		assert.NoError(t, err)
+		commitDone <- resp
+	}()
+
+	// commit must not be able to finalize the cache while upload is still
+	// mid-write and holding the lock -- this is exactly the window in which
+	// the pre-fix code let commit's cleanup delete the temp chunk upload was
+	// still writing to.
+	select {
+	case <-commitDone:
+		t.Fatal("commit finished while upload still held the cache lock")
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	close(releaseUpload)
+
+	uploadResp := <-uploadDone
+	require.NotNil(t, uploadResp)
+	defer uploadResp.Body.Close()
+	assert.Equal(t, 200, uploadResp.StatusCode)
+
+	commitResp := <-commitDone
+	require.NotNil(t, commitResp)
+	defer commitResp.Body.Close()
+	assert.Equal(t, 200, commitResp.StatusCode)
+
+	getResp, err := http.Get(fmt.Sprintf("%s/artifacts/%d", base, id))
+	require.NoError(t, err)
+	defer getResp.Body.Close()
+	require.Equal(t, 200, getResp.StatusCode)
+	got, err := io.ReadAll(getResp.Body)
+	require.NoError(t, err)
+	assert.Equal(t, content, got)
+}