Keep a tombstone on gem yank (#155)

Yank hard-deleted a local gem version, so the same name-version-platform could be re-pushed with different bytes and break lockfiles pinning the old checksum. rubygems.org forbids re-pushing a yanked version.

- write a `_yanked/<name>/<full>` tombstone before deleting the gem, quick spec and index line
- reject a push of a tombstoned version with 409
- serialise push and yank of one version on the engine Redis lock; 503 when it stays busy
- drop partial push writes on failure
- return 404 for `_index/` and `_yanked/` paths on gems locals
- encode out-of-range gemspec dates (before 1900) as the epoch instead of wrapping the Time#_dump year field
- extend the ruby:3.3 e2e to yank, re-push different bytes and fail to install the yanked version

Reviewed-on: #155
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>

internal/api/v1/gems.go

@@ -26,10 +26,17 @@ import (
 //   gems/<full>.gem                         package (immutable)
 //   quick/Marshal.4.8/<full>.gemspec.rz     marshalled spec (immutable)
 //   _index/<name>/<full>                    compact index /info line; gates visibility
+//   _yanked/<name>/<full>                   yank tombstone; blocks re-push of the version
 
-var gemsMaxPushBytes int64 = 256 << 20
+var (
+	gemsMaxPushBytes int64 = 256 << 20
+	gemsLockWait           = 30 * time.Second
+)
 
-const gemsIndexPrefix = "_index/"
+const (
+	gemsIndexPrefix  = "_index/"
+	gemsYankedPrefix = "_yanked/"
+)
 
 func gemsText(w http.ResponseWriter, status int, msg string) {
 	w.Header().Set("Content-Type", "text/plain; charset=utf-8")
@@ -46,6 +53,16 @@ func (h *ProxyHandler) localGemsRemote(w http.ResponseWriter, r *http.Request) (
 	return remote, true
 }
 
+// lockGemVersion serialises push and yank of one gem version across replicas.
+func (h *ProxyHandler) lockGemVersion(w http.ResponseWriter, r *http.Request, repo, full string) (func(), bool) {
+	unlock, err := h.engine.Lock(r.Context(), repo, "_gemlock/"+full, gemsLockWait, 5*time.Minute)
+	if err != nil {
+		gemsText(w, http.StatusServiceUnavailable, full+" is being pushed or yanked, retry")
+		return nil, false
+	}
+	return unlock, true
+}
+
 func (h *ProxyHandler) handleLocalPost(w http.ResponseWriter, r *http.Request) {
 	if chi.URLParam(r, "*") != "api/v1/gems" {
 		gemsText(w, http.StatusNotFound, "not found")
@@ -72,34 +89,61 @@ func (h *ProxyHandler) handleLocalPost(w http.ResponseWriter, r *http.Request) {
 	}
 
 	full := spec.FullName()
+	unlock, ok := h.lockGemVersion(w, r, remote.Name, full)
+	if !ok {
+		return
+	}
+	defer unlock()
 	conflict := func() {
 		gemsText(w, http.StatusConflict, fmt.Sprintf("%s already exists; overwrites are not allowed", full))
 	}
+	index := gemsIndexPrefix + spec.Name + "/" + full
+	for _, p := range []string{gemsYankedPrefix + spec.Name + "/" + full, index} {
+		f, err := h.db.GetLocalFile(r.Context(), remote.Name, p)
+		if err != nil {
+			gemsText(w, http.StatusInternalServerError, err.Error())
+			return
+		}
+		if f != nil && p == index {
+			conflict()
+			return
+		}
+		if f != nil {
+			gemsText(w, http.StatusConflict, fmt.Sprintf("%s was yanked; re-pushing a yanked version is not allowed", full))
+			return
+		}
+	}
+	// The version is not live, so on failure every path is an invisible leftover and safe to drop.
+	paths := []string{"gems/" + full + ".gem", "quick/Marshal.4.8/" + full + ".gemspec.rz", index}
+	fail := func(status int, msg string) {
+		for _, p := range paths {
+			if err := h.db.DeleteLocalFile(context.WithoutCancel(r.Context()), remote.Name, p); err != nil {
+				slog.Error("gem push cleanup failed", "repo", remote.Name, "path", p, "error", err)
+			}
+		}
+		gemsText(w, status, msg)
+	}
 	// The index line gates visibility, so it is written last.
 	for _, f := range []struct {
 		path string
 		data []byte
 	}{
-		{"gems/" + full + ".gem", gem},
-		{"quick/Marshal.4.8/" + full + ".gemspec.rz", quick},
+		{paths[0], gem},
+		{paths[1], quick},
 	} {
 		ok, err := h.npmStoreImmutable(r.Context(), remote.Name, f.path, f.data, "application/octet-stream")
 		if err != nil {
-			gemsText(w, http.StatusInternalServerError, err.Error())
+			fail(http.StatusInternalServerError, err.Error())
 			return
 		}
 		if !ok {
-			conflict()
+			fail(http.StatusConflict, fmt.Sprintf("%s already exists; overwrites are not allowed", full))
 			return
 		}
 	}
 	line := gems.InfoLine(spec, gem)
-	if _, err := h.npmStore(r.Context(), remote.Name, gemsIndexPrefix+spec.Name+"/"+full, []byte(line), "text/plain", false); err != nil {
-		if errors.Is(err, database.ErrAlreadyExists) {
-			conflict()
-			return
-		}
-		gemsText(w, http.StatusInternalServerError, err.Error())
+	if _, err := h.npmStore(r.Context(), remote.Name, index, []byte(line), "text/plain", false); err != nil {
+		fail(http.StatusInternalServerError, err.Error())
 		return
 	}
 
@@ -141,6 +185,11 @@ func (h *ProxyHandler) handleLocalDelete(w http.ResponseWriter, r *http.Request)
 	}
 
 	full := spec.FullName()
+	unlock, ok := h.lockGemVersion(w, r, remote.Name, full)
+	if !ok {
+		return
+	}
+	defer unlock()
 	index := gemsIndexPrefix + spec.Name + "/" + full
 	existing, err := h.db.GetLocalFile(r.Context(), remote.Name, index)
 	if err != nil {
@@ -151,8 +200,18 @@ func (h *ProxyHandler) handleLocalDelete(w http.ResponseWriter, r *http.Request)
 		gemsText(w, http.StatusNotFound, "The version "+spec.Version+" does not exist")
 		return
 	}
+	// The tombstone keeps the yanked line and is written before any delete.
+	line, err := h.readLocalBlob(r.Context(), existing.ContentHash)
+	if err != nil {
+		gemsText(w, http.StatusInternalServerError, err.Error())
+		return
+	}
+	if _, err := h.npmStore(r.Context(), remote.Name, gemsYankedPrefix+spec.Name+"/"+full, line, "text/plain", false); err != nil && !errors.Is(err, database.ErrAlreadyExists) {
+		gemsText(w, http.StatusInternalServerError, err.Error())
+		return
+	}
 	for _, p := range []string{index, "gems/" + full + ".gem", "quick/Marshal.4.8/" + full + ".gemspec.rz"} {
-		if err := h.db.DeleteLocalFile(r.Context(), remote.Name, p); err != nil {
+		if err := h.db.DeleteLocalFile(context.WithoutCancel(r.Context()), remote.Name, p); err != nil {
 			gemsText(w, http.StatusInternalServerError, err.Error())
 			return
 		}

internal/api/v1/gems_handler_test.go

@@ -4,14 +4,18 @@ import (
 	"archive/tar"
 	"bytes"
 	"compress/gzip"
+	"context"
 	"crypto/md5"
 	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
 	"net/http"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
+	"git.unkin.net/unkin/artifactapi/internal/cache"
 	"git.unkin.net/unkin/artifactapi/pkg/models"
 )
 
@@ -134,8 +138,26 @@ func TestGemsPushIndexYank(t *testing.T) {
 	if code, _ := npmDo(t, http.MethodGet, base+"gems/hello_gem-0.2.0-java.gem", ""); code != http.StatusNotFound {
 		t.Fatalf("yanked download = %d, want 404", code)
 	}
-	if code, _ := npmDo(t, http.MethodPost, base+"api/v1/gems", string(gj)); code != http.StatusOK {
-		t.Fatalf("push after yank = %d", code)
+	if _, v := npmDo(t, http.MethodGet, base+"versions", ""); strings.Contains(v, "java") || !strings.Contains(v, "hello_gem 0.1.0,0.2.0 ") {
+		t.Fatalf("versions after yank = %q", v)
+	}
+	for _, g := range []string{string(gj), string(gj) + "x"} {
+		if code, body := npmDo(t, http.MethodPost, base+"api/v1/gems", g); code != http.StatusConflict || !strings.Contains(body, "yanked") {
+			t.Fatalf("push after yank = %d %q, want 409", code, body)
+		}
+	}
+	if _, info := npmDo(t, http.MethodGet, base+"info/hello_gem", ""); strings.Contains(info, "java") {
+		t.Fatalf("info after rejected re-push = %q", info)
+	}
+
+	if c := yank("gem_name=hello-gem&version=1.0.0"); c != http.StatusOK {
+		t.Fatalf("yank last version = %d", c)
+	}
+	if code, _ := npmDo(t, http.MethodGet, base+"info/hello-gem", ""); code != http.StatusNotFound {
+		t.Fatalf("info for fully yanked gem = %d, want 404", code)
+	}
+	if _, names := npmDo(t, http.MethodGet, base+"names", ""); names != "---\nhello_gem\n" {
+		t.Fatalf("names after yanking every version = %q", names)
 	}
 }
 
@@ -157,3 +179,75 @@ func TestGemsPushRejects(t *testing.T) {
 		t.Fatalf("oversized push = %d, want 413", code)
 	}
 }
+
+func TestGemsPushYankSerialised(t *testing.T) {
+	h, srv := npmTestServer(t)
+	repo := npmRepo(t, h, models.PackageGems)
+	base := srv.URL + "/api/v1/local/" + repo + "/"
+	g := string(testGemFile(t, "race", "1.0.0", "ruby"))
+	var wg sync.WaitGroup
+	for i := 0; i < 16; i++ {
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			if code, body := npmDo(t, http.MethodPost, base+"api/v1/gems", g); code != http.StatusOK && code != http.StatusConflict {
+				t.Errorf("push = %d: %s", code, body)
+			}
+		}()
+		go func() {
+			defer wg.Done()
+			if code, body := npmDo(t, http.MethodDelete, base+"api/v1/gems/yank?gem_name=race&version=1.0.0", ""); code != http.StatusOK && code != http.StatusNotFound {
+				t.Errorf("yank = %d: %s", code, body)
+			}
+		}()
+	}
+	wg.Wait()
+
+	present := func(p string) bool {
+		f, err := h.db.GetLocalFile(context.Background(), repo, p)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return f != nil
+	}
+	live := []string{"_index/race/race-1.0.0", "gems/race-1.0.0.gem", "quick/Marshal.4.8/race-1.0.0.gemspec.rz"}
+	tomb := present("_yanked/race/race-1.0.0")
+	for _, p := range live {
+		if present(p) == tomb {
+			t.Errorf("tombstoned=%v but %s present=%v", tomb, p, !tomb)
+		}
+	}
+	_, info := npmDo(t, http.MethodGet, base+"info/race", "")
+	if strings.Contains(info, "1.0.0") == tomb {
+		t.Errorf("tombstoned=%v but info = %q", tomb, info)
+	}
+}
+
+func TestGemsLockTimeoutIs503(t *testing.T) {
+	h, srv := npmTestServer(t)
+	repo := npmRepo(t, h, models.PackageGems)
+	base := srv.URL + "/api/v1/local/" + repo + "/"
+	redis, err := cache.NewRedis(testRedis)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if ok, err := redis.AcquireLock(context.Background(), repo, "_gemlock/busy-1.0.0", time.Minute); !ok || err != nil {
+		t.Fatalf("hold lock = %v %v", ok, err)
+	}
+	prev := gemsLockWait
+	gemsLockWait = 100 * time.Millisecond
+	t.Cleanup(func() { gemsLockWait = prev })
+
+	if code, _ := npmDo(t, http.MethodPost, base+"api/v1/gems", string(testGemFile(t, "busy", "1.0.0", "ruby"))); code != http.StatusServiceUnavailable {
+		t.Fatalf("push under held lock = %d, want 503", code)
+	}
+	if code, _ := npmDo(t, http.MethodDelete, base+"api/v1/gems/yank?gem_name=busy&version=1.0.0", ""); code != http.StatusServiceUnavailable {
+		t.Fatalf("yank under held lock = %d, want 503", code)
+	}
+	if err := redis.ReleaseLock(context.Background(), repo, "_gemlock/busy-1.0.0"); err != nil {
+		t.Fatal(err)
+	}
+	if code, body := npmDo(t, http.MethodPost, base+"api/v1/gems", string(testGemFile(t, "busy", "1.0.0", "ruby"))); code != http.StatusOK {
+		t.Fatalf("push after release = %d: %s", code, body)
+	}
+}

internal/api/v1/npm_handler_test.go

@@ -15,7 +15,9 @@ import (
 
 	"github.com/go-chi/chi/v5"
 
+	"git.unkin.net/unkin/artifactapi/internal/cache"
 	"git.unkin.net/unkin/artifactapi/internal/database"
+	"git.unkin.net/unkin/artifactapi/internal/proxy"
 	"git.unkin.net/unkin/artifactapi/internal/storage"
 	"git.unkin.net/unkin/artifactapi/internal/testsupport"
 	"git.unkin.net/unkin/artifactapi/internal/virtual"
@@ -25,6 +27,7 @@ import (
 var (
 	testDSN   string
 	testMinio testsupport.MinioConn
+	testRedis string
 )
 
 func TestMain(m *testing.M) {
@@ -38,8 +41,15 @@ func TestMain(m *testing.M) {
 		termPG()
 		os.Exit(m.Run())
 	}
-	testDSN, testMinio = dsn, conn
+	redisURL, termRedis, err := testsupport.StartRedis(ctx)
+	if err != nil {
+		termMinio()
+		termPG()
+		os.Exit(m.Run())
+	}
+	testDSN, testMinio, testRedis = dsn, conn, redisURL
 	code := m.Run()
+	termRedis()
 	termMinio()
 	termPG()
 	os.Exit(code)
@@ -65,7 +75,11 @@ func npmTestServer(t *testing.T) (*ProxyHandler, *httptest.Server) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	h := NewProxyHandler(nil, nil, db, store, nil)
+	redis, err := cache.NewRedis(testRedis)
+	if err != nil {
+		t.Fatal(err)
+	}
+	h := NewProxyHandler(proxy.NewEngine(db, redis, store), nil, db, store, nil)
 	r := chi.NewRouter()
 	r.Mount("/api/v1", h.Routes())
 	srv := httptest.NewServer(r)

internal/api/v1/proxy.go

@@ -263,9 +263,15 @@ func (h *ProxyHandler) handleLocal(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 	}
-	if remote.PackageType == models.PackageGems && isGemsIndexPath(path) {
-		h.serveLocalGemsIndex(w, r, remote, path)
-		return
+	if remote.PackageType == models.PackageGems {
+		if isGemsIndexPath(path) {
+			h.serveLocalGemsIndex(w, r, remote, path)
+			return
+		}
+		if strings.HasPrefix(path, gemsIndexPrefix) || strings.HasPrefix(path, gemsYankedPrefix) {
+			http.Error(w, "not found", http.StatusNotFound)
+			return
+		}
 	}
 
 	prov, _ := provider.Get(remote.PackageType)