From 4a6bd76976a56c44fed48a341948f298dd8bbab7 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 18 Oct 2026 01:24:44 +1100 Subject: [PATCH 1/4] Keep a tombstone on gem yank and reject re-pushing yanked versions --- e2e-docker/README.md | 3 ++- e2e-docker/gems_local_test.go | 15 ++++++++++----- internal/api/v1/gems.go | 25 ++++++++++++++++++++++++- internal/api/v1/gems_handler_test.go | 22 ++++++++++++++++++++-- internal/provider/gems/local.go | 7 ++++--- internal/provider/gems/local_test.go | 9 +++++++++ 6 files changed, 69 insertions(+), 12 deletions(-) diff --git a/e2e-docker/README.md b/e2e-docker/README.md index f2c9092..ca96aab 100644 --- a/e2e-docker/README.md +++ b/e2e-docker/README.md @@ -30,7 +30,8 @@ already-running stack. index), rpm (real package + **automatic repodata** generation), npm (stock `node:22` client: `npm publish` unscoped + scoped, `npm install` + `require()`, second version, republish rejected), gems (stock `ruby:3.3`: `gem push`, - `gem install`, `bundle install` + `require`, republish rejected, `gem yank`). + `gem install`, `bundle install` + `require`, republish rejected, `gem yank`, + yanked re-push rejected). - **Virtual repositories** — pypi simple-index merge and helm `index.yaml` merge across two members; npm packument merge over two locals via a stock `node:22` client (merged versions, install through the member redirect). diff --git a/e2e-docker/gems_local_test.go b/e2e-docker/gems_local_test.go index b5f5134..17d1068 100644 --- a/e2e-docker/gems_local_test.go +++ b/e2e-docker/gems_local_test.go @@ -13,7 +13,8 @@ import ( // TestLocalGemsPushInstall drives stock gem and bundler clients against a // local gems repo: push two versions, reject a republish, gem install and -// bundle install the latest via the compact index, then yank it. +// bundle install the latest via the compact index, then yank it and check a +// re-push of the yanked version with different bytes is rejected. func TestLocalGemsPushInstall(t *testing.T) { network := os.Getenv("COMPOSE_NETWORK") internal := os.Getenv("ARTIFACTAPI_INTERNAL") @@ -31,8 +32,8 @@ func TestLocalGemsPushInstall(t *testing.T) { src := strings.TrimRight(internal, "/") + "/api/v1/local/" + name script := `set -euo pipefail export GEM_HOST_API_KEY=unused -mk() { d=/w/src-$1; mkdir -p $d/lib; cd $d - echo "module UnkinE2eHello; VERSION = '$1'; end" > lib/unkin-e2e-hello.rb +mk() { d=/w/src-$1${2:+-b}; mkdir -p $d/lib; cd $d + echo "module UnkinE2eHello; VERSION = '$1'; end${2:-}" > lib/unkin-e2e-hello.rb printf 'Gem::Specification.new { |s| s.name = "unkin-e2e-hello"; s.version = "%s"; s.summary = "e2e"; s.authors = ["e2e"]; s.files = ["lib/unkin-e2e-hello.rb"] }' "$1" > g.gemspec gem build g.gemspec >/dev/null 2>&1; } mk 0.1.0 && gem push --host "$SRC" unkin-e2e-hello-0.1.0.gem @@ -47,14 +48,18 @@ printf 'source "%s/"\ngem "unkin-e2e-hello"\n' "$SRC" > Gemfile bundle install bundle exec ruby -e 'require "unkin-e2e-hello"; puts "BUNDLE #{UnkinE2eHello::VERSION}"' gem yank unkin-e2e-hello -v 0.2.0 --host "$SRC" -echo "INFO $(curl -s "$SRC/info/unkin-e2e-hello" | cut -d' ' -f1 | tr '\n' ' ')"` +echo "INFO $(curl -s "$SRC/info/unkin-e2e-hello" | cut -d' ' -f1 | tr '\n' ' ')" +mk 0.2.0 '# changed' && if gem push --host "$SRC" unkin-e2e-hello-0.2.0.gem; then echo "yanked re-push accepted"; exit 1; fi +cd /w/app && gem uninstall -a -x unkin-e2e-hello >/dev/null +if gem install --remote --no-document unkin-e2e-hello -v 0.2.0; then echo "yanked version installed"; exit 1; fi +echo "VERSIONS $(curl -s "$SRC/versions" | grep '^unkin-e2e-hello ' | cut -d' ' -f2)"` out, err := exec.Command("docker", "run", "--rm", "--network", network, "-e", "SRC="+src, "ruby:3.3", "bash", "-c", script).CombinedOutput() if err != nil { t.Fatalf("gem client run failed: %v\n%s", err, out) } - for _, want := range []string{"GEM 0.2.0", "BUNDLE 0.2.0", "INFO --- 0.1.0 "} { + for _, want := range []string{"GEM 0.2.0", "BUNDLE 0.2.0", "INFO --- 0.1.0 ", "re-pushing a yanked version is not allowed", "VERSIONS 0.1.0\n"} { if !strings.Contains(string(out), want) { t.Fatalf("missing %q in gem output:\n%s", want, out) } diff --git a/internal/api/v1/gems.go b/internal/api/v1/gems.go index ba64927..3ce02b6 100644 --- a/internal/api/v1/gems.go +++ b/internal/api/v1/gems.go @@ -26,10 +26,14 @@ import ( // gems/.gem package (immutable) // quick/Marshal.4.8/.gemspec.rz marshalled spec (immutable) // _index// compact index /info line; gates visibility +// _yanked// yank tombstone; blocks re-push of the version var gemsMaxPushBytes int64 = 256 << 20 -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") @@ -75,6 +79,15 @@ func (h *ProxyHandler) handleLocalPost(w http.ResponseWriter, r *http.Request) { conflict := func() { gemsText(w, http.StatusConflict, fmt.Sprintf("%s already exists; overwrites are not allowed", full)) } + yanked, err := h.db.GetLocalFile(r.Context(), remote.Name, gemsYankedPrefix+spec.Name+"/"+full) + if err != nil { + gemsText(w, http.StatusInternalServerError, err.Error()) + return + } + if yanked != nil { + gemsText(w, http.StatusConflict, fmt.Sprintf("%s was yanked; re-pushing a yanked version is not allowed", full)) + return + } // The index line gates visibility, so it is written last. for _, f := range []struct { path string @@ -151,6 +164,16 @@ 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 { gemsText(w, http.StatusInternalServerError, err.Error()) diff --git a/internal/api/v1/gems_handler_test.go b/internal/api/v1/gems_handler_test.go index 095db07..fcbd6d6 100644 --- a/internal/api/v1/gems_handler_test.go +++ b/internal/api/v1/gems_handler_test.go @@ -134,8 +134,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) } } diff --git a/internal/provider/gems/local.go b/internal/provider/gems/local.go index 6422e8f..0cbb54f 100644 --- a/internal/provider/gems/local.go +++ b/internal/provider/gems/local.go @@ -445,10 +445,11 @@ func (m *rbMarshal) requirement(r Requirement) { // time writes a UTC Time with day resolution, as Time#_dump lays it out. func (m *rbMarshal) time(t time.Time) { - if t.IsZero() { - t = time.Unix(0, 0) - } t = t.UTC() + // Time#_dump holds year-1900 in 16 bits. + if t.IsZero() || t.Year() < 1900 || t.Year() > 1900+0xffff { + t = time.Unix(0, 0).UTC() + } p := uint32(1)<<31 | uint32(1)<<30 | uint32(t.Year()-1900)<<14 | uint32(t.Month()-1)<<10 | uint32(t.Day())<<5 | uint32(t.Hour()) s := uint32(t.Minute())<<26 | uint32(t.Second())<<20 m.WriteString("Iu") diff --git a/internal/provider/gems/local_test.go b/internal/provider/gems/local_test.go index 9bcf685..4df4f54 100644 --- a/internal/provider/gems/local_test.go +++ b/internal/provider/gems/local_test.go @@ -175,6 +175,15 @@ func TestRubyMarshalEncoding(t *testing.T) { if want := "Iu:\x09Time\x0d\x20\xa6\x1f\xc0\x00\x00\x00\x00\x06:\x09zoneI\"\x08UTC\x06:\x06EF"; m.String() != want { t.Errorf("time = %q, want %q", m.String(), want) } + var epoch rbMarshal + epoch.time(time.Time{}) + for _, y := range []int{1, 1899, 1900 + 0x10000} { + m.Reset() + m.time(time.Date(y, 1, 1, 0, 0, 0, 0, time.UTC)) + if m.String() != epoch.String() { + t.Errorf("time(year %d) = %q, want the epoch encoding %q", y, m.String(), epoch.String()) + } + } } func TestQuickSpec(t *testing.T) { -- 2.47.3 From 960a5e09a3d9d9a01a8ab59a3cf2bd2732bc386d Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 18 Oct 2026 01:33:35 +1100 Subject: [PATCH 2/4] Roll back a gem push that races a yank and hide gems bookkeeping paths --- internal/api/v1/gems.go | 42 ++++++++++++++++++++-------- internal/api/v1/gems_handler_test.go | 26 +++++++++++++++++ internal/api/v1/proxy.go | 12 ++++++-- 3 files changed, 66 insertions(+), 14 deletions(-) diff --git a/internal/api/v1/gems.go b/internal/api/v1/gems.go index 3ce02b6..5d81de8 100644 --- a/internal/api/v1/gems.go +++ b/internal/api/v1/gems.go @@ -28,7 +28,10 @@ import ( // _index// compact index /info line; gates visibility // _yanked// yank tombstone; blocks re-push of the version -var gemsMaxPushBytes int64 = 256 << 20 +var ( + gemsMaxPushBytes int64 = 256 << 20 + gemsPushHook = func() {} +) const ( gemsIndexPrefix = "_index/" @@ -79,22 +82,30 @@ func (h *ProxyHandler) handleLocalPost(w http.ResponseWriter, r *http.Request) { conflict := func() { gemsText(w, http.StatusConflict, fmt.Sprintf("%s already exists; overwrites are not allowed", full)) } - yanked, err := h.db.GetLocalFile(r.Context(), remote.Name, gemsYankedPrefix+spec.Name+"/"+full) - if err != nil { - gemsText(w, http.StatusInternalServerError, err.Error()) - return - } - if yanked != nil { - gemsText(w, http.StatusConflict, fmt.Sprintf("%s was yanked; re-pushing a yanked version is not allowed", full)) + index := gemsIndexPrefix + spec.Name + "/" + full + paths := []string{index, "gems/" + full + ".gem", "quick/Marshal.4.8/" + full + ".gemspec.rz"} + yanked := func() bool { + f, err := h.db.GetLocalFile(r.Context(), remote.Name, gemsYankedPrefix+spec.Name+"/"+full) + if err != nil { + gemsText(w, http.StatusInternalServerError, err.Error()) + return true + } + if f != nil { + gemsText(w, http.StatusConflict, fmt.Sprintf("%s was yanked; re-pushing a yanked version is not allowed", full)) + } + return f != nil + } + if yanked() { return } + gemsPushHook() // 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[1], gem}, + {paths[2], quick}, } { ok, err := h.npmStoreImmutable(r.Context(), remote.Name, f.path, f.data, "application/octet-stream") if err != nil { @@ -107,7 +118,7 @@ func (h *ProxyHandler) handleLocalPost(w http.ResponseWriter, r *http.Request) { } } 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 _, err := h.npmStore(r.Context(), remote.Name, index, []byte(line), "text/plain", false); err != nil { if errors.Is(err, database.ErrAlreadyExists) { conflict() return @@ -115,6 +126,15 @@ func (h *ProxyHandler) handleLocalPost(w http.ResponseWriter, r *http.Request) { gemsText(w, http.StatusInternalServerError, err.Error()) return } + // A yank that raced this push tombstoned the version before or after its deletes ran; undo our writes. + if yanked() { + for _, p := range paths { + if err := h.db.DeleteLocalFile(r.Context(), remote.Name, p); err != nil { + slog.Error("gem push rollback failed", "repo", remote.Name, "path", p, "error", err) + } + } + return + } slog.Info("local gem pushed", "repo", remote.Name, "gem", full) gemsText(w, http.StatusOK, "Successfully registered gem: "+spec.Name+" ("+spec.Version+")") diff --git a/internal/api/v1/gems_handler_test.go b/internal/api/v1/gems_handler_test.go index fcbd6d6..1d56e9b 100644 --- a/internal/api/v1/gems_handler_test.go +++ b/internal/api/v1/gems_handler_test.go @@ -175,3 +175,29 @@ func TestGemsPushRejects(t *testing.T) { t.Fatalf("oversized push = %d, want 413", code) } } + +func TestGemsPushRacingYank(t *testing.T) { + h, srv := npmTestServer(t) + repo := npmRepo(t, h, models.PackageGems) + base := srv.URL + "/api/v1/local/" + repo + "/" + g := testGemFile(t, "race", "1.0.0", "ruby") + if code, body := npmDo(t, http.MethodPost, base+"api/v1/gems", string(g)); code != http.StatusOK { + t.Fatalf("push = %d: %s", code, body) + } + + gemsPushHook = func() { + gemsPushHook = func() {} + if code, body := npmDo(t, http.MethodDelete, base+"api/v1/gems/yank?gem_name=race&version=1.0.0", ""); code != http.StatusOK { + t.Errorf("yank = %d: %s", code, body) + } + } + t.Cleanup(func() { gemsPushHook = func() {} }) + if code, body := npmDo(t, http.MethodPost, base+"api/v1/gems", string(g)); code != http.StatusConflict || !strings.Contains(body, "yanked") { + t.Fatalf("push racing yank = %d %q, want 409", code, body) + } + for _, p := range []string{"info/race", "gems/race-1.0.0.gem", "quick/Marshal.4.8/race-1.0.0.gemspec.rz", "_index/race/race-1.0.0", "_yanked/race/race-1.0.0"} { + if code, _ := npmDo(t, http.MethodGet, base+p, ""); code != http.StatusNotFound { + t.Errorf("GET %s = %d, want 404", p, code) + } + } +} diff --git a/internal/api/v1/proxy.go b/internal/api/v1/proxy.go index 16ebc55..4c2f11f 100644 --- a/internal/api/v1/proxy.go +++ b/internal/api/v1/proxy.go @@ -257,9 +257,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) -- 2.47.3 From d0fd53222cf020be2c6242c1d9d8447c6626da72 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 18 Oct 2026 01:44:20 +1100 Subject: [PATCH 3/4] Serialise gem push and yank per version with a Redis lock --- internal/api/v1/gems.go | 72 ++++++++++++++---------- internal/api/v1/gems_handler_test.go | 82 ++++++++++++++++++++++------ internal/api/v1/npm_handler_test.go | 18 +++++- internal/proxy/engine.go | 24 +++++--- 4 files changed, 142 insertions(+), 54 deletions(-) diff --git a/internal/api/v1/gems.go b/internal/api/v1/gems.go index 5d81de8..912777f 100644 --- a/internal/api/v1/gems.go +++ b/internal/api/v1/gems.go @@ -30,7 +30,7 @@ import ( var ( gemsMaxPushBytes int64 = 256 << 20 - gemsPushHook = func() {} + gemsLockWait = 30 * time.Second ) const ( @@ -53,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") @@ -79,60 +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 - paths := []string{index, "gems/" + full + ".gem", "quick/Marshal.4.8/" + full + ".gemspec.rz"} - yanked := func() bool { - f, err := h.db.GetLocalFile(r.Context(), remote.Name, gemsYankedPrefix+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 true + 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 } - return f != nil } - if yanked() { - 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) } - gemsPushHook() // The index line gates visibility, so it is written last. for _, f := range []struct { path string data []byte }{ - {paths[1], gem}, - {paths[2], 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, index, []byte(line), "text/plain", false); err != nil { - if errors.Is(err, database.ErrAlreadyExists) { - conflict() - return - } - gemsText(w, http.StatusInternalServerError, err.Error()) - return - } - // A yank that raced this push tombstoned the version before or after its deletes ran; undo our writes. - if yanked() { - for _, p := range paths { - if err := h.db.DeleteLocalFile(r.Context(), remote.Name, p); err != nil { - slog.Error("gem push rollback failed", "repo", remote.Name, "path", p, "error", err) - } - } + fail(http.StatusInternalServerError, err.Error()) return } @@ -174,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 { @@ -195,7 +211,7 @@ func (h *ProxyHandler) handleLocalDelete(w http.ResponseWriter, r *http.Request) 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 } diff --git a/internal/api/v1/gems_handler_test.go b/internal/api/v1/gems_handler_test.go index 1d56e9b..110b735 100644 --- a/internal/api/v1/gems_handler_test.go +++ b/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" ) @@ -176,28 +180,74 @@ func TestGemsPushRejects(t *testing.T) { } } -func TestGemsPushRacingYank(t *testing.T) { +func TestGemsPushYankSerialised(t *testing.T) { h, srv := npmTestServer(t) repo := npmRepo(t, h, models.PackageGems) base := srv.URL + "/api/v1/local/" + repo + "/" - g := testGemFile(t, "race", "1.0.0", "ruby") - if code, body := npmDo(t, http.MethodPost, base+"api/v1/gems", string(g)); code != http.StatusOK { - t.Fatalf("push = %d: %s", code, body) + 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() - gemsPushHook = func() { - gemsPushHook = func() {} - if code, body := npmDo(t, http.MethodDelete, base+"api/v1/gems/yank?gem_name=race&version=1.0.0", ""); code != http.StatusOK { - t.Errorf("yank = %d: %s", code, body) + 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) } } - t.Cleanup(func() { gemsPushHook = func() {} }) - if code, body := npmDo(t, http.MethodPost, base+"api/v1/gems", string(g)); code != http.StatusConflict || !strings.Contains(body, "yanked") { - t.Fatalf("push racing yank = %d %q, want 409", code, body) - } - for _, p := range []string{"info/race", "gems/race-1.0.0.gem", "quick/Marshal.4.8/race-1.0.0.gemspec.rz", "_index/race/race-1.0.0", "_yanked/race/race-1.0.0"} { - if code, _ := npmDo(t, http.MethodGet, base+p, ""); code != http.StatusNotFound { - t.Errorf("GET %s = %d, want 404", p, code) - } + _, 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) } } diff --git a/internal/api/v1/npm_handler_test.go b/internal/api/v1/npm_handler_test.go index 95c8f3b..695efe9 100644 --- a/internal/api/v1/npm_handler_test.go +++ b/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) diff --git a/internal/proxy/engine.go b/internal/proxy/engine.go index 1a6ca2f..d097733 100644 --- a/internal/proxy/engine.go +++ b/internal/proxy/engine.go @@ -228,10 +228,11 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error { return fmt.Errorf("get remote: %w", err) } if !wildcard { - if err := e.waitForLock(ctx, remoteName, path); err != nil { + release, err := e.Lock(ctx, remoteName, path, e.evictLockWait, fetchLockTTL) + if err != nil { return err } - defer func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path) }() + defer release() if err := e.db.DeleteArtifact(ctx, remoteName, path); err != nil { return fmt.Errorf("delete artifact: %w", err) } @@ -252,13 +253,20 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error { return e.cache.ForgetPrefix(ctx, remoteName, prefix) } -// waitForLock takes the per-path fetch lock so an in-flight Fetch cannot -// re-set TTL/ETag keys after an evict. It fails with a 503 when the lock -// cannot be taken within evictLockWait or Redis errors. -func (e *Engine) waitForLock(ctx context.Context, remoteName, path string) error { - deadline := time.Now().Add(e.evictLockWait) +// Lock takes the per-path Redis lock (the one Fetch holds) and returns its +// release. It fails with a 503 when the lock cannot be taken within wait or +// Redis errors. +func (e *Engine) Lock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) (func(), error) { + if err := e.waitForLock(ctx, remoteName, path, wait, ttl); err != nil { + return nil, err + } + return func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path) }, nil +} + +func (e *Engine) waitForLock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) error { + deadline := time.Now().Add(wait) for { - ok, err := e.cache.AcquireLock(ctx, remoteName, path, fetchLockTTL) + ok, err := e.cache.AcquireLock(ctx, remoteName, path, ttl) if ok { return nil } -- 2.47.3 From a431319dab2bd796b61feda13ce8ada225710c69 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 18 Oct 2026 02:14:07 +1100 Subject: [PATCH 4/4] Release Redis locks only while the caller still owns them --- internal/api/v1/gems_handler_test.go | 7 ++-- internal/cache/cache_test.go | 50 ++++++++++++++++++++++++---- internal/cache/redis.go | 19 ++++++++--- internal/proxy/engine.go | 26 ++++++++------- internal/proxy/evict_test.go | 20 ++++++----- 5 files changed, 86 insertions(+), 36 deletions(-) diff --git a/internal/api/v1/gems_handler_test.go b/internal/api/v1/gems_handler_test.go index 110b735..c1cc0cc 100644 --- a/internal/api/v1/gems_handler_test.go +++ b/internal/api/v1/gems_handler_test.go @@ -231,8 +231,9 @@ func TestGemsLockTimeoutIs503(t *testing.T) { 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) + token, err := redis.AcquireLock(context.Background(), repo, "_gemlock/busy-1.0.0", time.Minute) + if token == "" || err != nil { + t.Fatalf("hold lock = %q %v", token, err) } prev := gemsLockWait gemsLockWait = 100 * time.Millisecond @@ -244,7 +245,7 @@ func TestGemsLockTimeoutIs503(t *testing.T) { 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 { + if err := redis.ReleaseLock(context.Background(), repo, "_gemlock/busy-1.0.0", token); 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 { diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 8bcb380..a1e1dd3 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -61,18 +61,54 @@ func TestTTL(t *testing.T) { func TestLock(t *testing.T) { requireRedis(t) ctx := context.Background() - ok, err := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute) - if err != nil || !ok { - t.Fatalf("first acquire should succeed: %v %v", ok, err) + token, err := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute) + if err != nil || token == "" { + t.Fatalf("first acquire should succeed: %q %v", token, err) } - if ok, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); ok { + if again, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); again != "" { t.Error("second acquire should fail while held") } - if err := testRedis.ReleaseLock(ctx, "r", "lockpath"); err != nil { + if err := testRedis.ReleaseLock(ctx, "r", "lockpath", "not-the-owner"); err != nil { t.Fatal(err) } - if ok, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); !ok { - t.Error("acquire should succeed after release") + if again, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); again != "" { + t.Error("release with a foreign token must not drop the lock") + } + if err := testRedis.ReleaseLock(ctx, "r", "lockpath", token); err != nil { + t.Fatal(err) + } + next, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute) + if next == "" { + t.Fatal("acquire should succeed after release") + } + if next == token { + t.Error("tokens must differ between holders") + } +} + +func TestStaleReleaseKeepsNextHolder(t *testing.T) { + requireRedis(t) + ctx := context.Background() + stale, err := testRedis.AcquireLock(ctx, "r", "stale", 50*time.Millisecond) + if err != nil || stale == "" { + t.Fatalf("acquire: %q %v", stale, err) + } + time.Sleep(100 * time.Millisecond) + fresh, err := testRedis.AcquireLock(ctx, "r", "stale", time.Minute) + if err != nil || fresh == "" { + t.Fatalf("acquire after expiry: %q %v", fresh, err) + } + if err := testRedis.ReleaseLock(ctx, "r", "stale", stale); err != nil { + t.Fatal(err) + } + if again, _ := testRedis.AcquireLock(ctx, "r", "stale", time.Minute); again != "" { + t.Fatal("stale holder's release dropped the second holder's lock") + } + if err := testRedis.ReleaseLock(ctx, "r", "stale", fresh); err != nil { + t.Fatal(err) + } + if again, _ := testRedis.AcquireLock(ctx, "r", "stale", time.Minute); again == "" { + t.Error("owner release should free the lock") } } diff --git a/internal/cache/redis.go b/internal/cache/redis.go index eeca6c5..c120745 100644 --- a/internal/cache/redis.go +++ b/internal/cache/redis.go @@ -2,6 +2,7 @@ package cache import ( "context" + "crypto/rand" "fmt" "strings" "time" @@ -46,15 +47,23 @@ func (r *Redis) CheckTTL(ctx context.Context, remote, path string) (bool, error) return exists > 0, nil } -func (r *Redis) AcquireLock(ctx context.Context, remote, path string, ttl time.Duration) (bool, error) { +// AcquireLock returns the owner token on success and "" when the lock is held. +func (r *Redis) AcquireLock(ctx context.Context, remote, path string, ttl time.Duration) (string, error) { key := fmt.Sprintf("lock:%s:%s", remote, path) - ok, err := r.client.SetNX(ctx, key, "1", ttl).Result() - return ok, err + token := rand.Text() + ok, err := r.client.SetNX(ctx, key, token, ttl).Result() + if err != nil || !ok { + return "", err + } + return token, nil } -func (r *Redis) ReleaseLock(ctx context.Context, remote, path string) error { +var releaseScript = redis.NewScript(`if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end`) + +// ReleaseLock deletes the lock only while token still owns it. +func (r *Redis) ReleaseLock(ctx context.Context, remote, path, token string) error { key := fmt.Sprintf("lock:%s:%s", remote, path) - return r.client.Del(ctx, key).Err() + return releaseScript.Run(ctx, r.client, []string{key}, token).Err() } // SetETag stores etag for path, or drops the stored one when etag is empty. diff --git a/internal/proxy/engine.go b/internal/proxy/engine.go index d097733..99d24b7 100644 --- a/internal/proxy/engine.go +++ b/internal/proxy/engine.go @@ -134,10 +134,11 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p slog.Warn("cache hit but S3 miss, re-fetching", "remote", remote.Name, "path", path) } - locked, err := e.cache.AcquireLock(ctx, remote.Name, path, fetchLockTTL) + token, err := e.cache.AcquireLock(ctx, remote.Name, path, fetchLockTTL) if err != nil { slog.Warn("lock acquire failed", "error", err) } + locked := token != "" if !locked { // Another request holds the fetch lock. Poll the store until the leader @@ -151,7 +152,7 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p } if locked { - defer e.cache.ReleaseLock(ctx, remote.Name, path) + defer func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remote.Name, path, token) }() } if class == ClassMutable && remote.CheckMutable { @@ -257,31 +258,32 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error { // release. It fails with a 503 when the lock cannot be taken within wait or // Redis errors. func (e *Engine) Lock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) (func(), error) { - if err := e.waitForLock(ctx, remoteName, path, wait, ttl); err != nil { + token, err := e.waitForLock(ctx, remoteName, path, wait, ttl) + if err != nil { return nil, err } - return func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path) }, nil + return func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path, token) }, nil } -func (e *Engine) waitForLock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) error { +func (e *Engine) waitForLock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) (string, error) { deadline := time.Now().Add(wait) for { - ok, err := e.cache.AcquireLock(ctx, remoteName, path, ttl) - if ok { - return nil + token, err := e.cache.AcquireLock(ctx, remoteName, path, ttl) + if token != "" { + return token, nil } if ctx.Err() != nil { - return ctx.Err() + return "", ctx.Err() } if err != nil { - return &ProxyError{Status: http.StatusServiceUnavailable, Message: fmt.Sprintf("fetch lock: %v", err)} + return "", &ProxyError{Status: http.StatusServiceUnavailable, Message: fmt.Sprintf("lock: %v", err)} } if time.Now().After(deadline) { - return &ProxyError{Status: http.StatusServiceUnavailable, Message: "fetch in progress, retry evict"} + return "", &ProxyError{Status: http.StatusServiceUnavailable, Message: "resource busy, retry later"} } select { case <-ctx.Done(): - return ctx.Err() + return "", ctx.Err() case <-time.After(50 * time.Millisecond): } } diff --git a/internal/proxy/evict_test.go b/internal/proxy/evict_test.go index 622cc35..55635f4 100644 --- a/internal/proxy/evict_test.go +++ b/internal/proxy/evict_test.go @@ -152,13 +152,15 @@ func TestEvictUnknownRemote(t *testing.T) { } } -func holdLock(t *testing.T, remote, path string) { +func holdLock(t *testing.T, remote, path string) string { t.Helper() ctx := context.Background() - if ok, err := testCache.AcquireLock(ctx, remote, path, time.Minute); !ok || err != nil { - t.Fatalf("acquire: %v %v", ok, err) + token, err := testCache.AcquireLock(ctx, remote, path, time.Minute) + if token == "" || err != nil { + t.Fatalf("acquire: %q %v", token, err) } - t.Cleanup(func() { _ = testCache.ReleaseLock(ctx, remote, path) }) + t.Cleanup(func() { _ = testCache.ReleaseLock(ctx, remote, path, token) }) + return token } func TestEvictWaitsForFetchLock(t *testing.T) { @@ -167,7 +169,7 @@ func TestEvictWaitsForFetchLock(t *testing.T) { r := rpmRemote(t, "evict-lock", srv.URL) ctx := context.Background() const path = "8/Everything/x86_64/repodata/repomd.xml" - holdLock(t, r.Name, path) + token := holdLock(t, r.Name, path) done := make(chan error, 1) go func() { done <- testEngine.Evict(ctx, r.Name, path) }() select { @@ -175,13 +177,13 @@ func TestEvictWaitsForFetchLock(t *testing.T) { t.Fatalf("evict returned while fetch lock held: %v", err) case <-time.After(200 * time.Millisecond): } - _ = testCache.ReleaseLock(ctx, r.Name, path) + _ = testCache.ReleaseLock(ctx, r.Name, path, token) if err := <-done; err != nil { t.Fatalf("evict: %v", err) } - ok, err := testCache.AcquireLock(ctx, r.Name, path, time.Second) - if !ok || err != nil { - t.Fatalf("lock still held after evict: %v %v", ok, err) + token, err := testCache.AcquireLock(ctx, r.Name, path, time.Second) + if token == "" || err != nil { + t.Fatalf("lock still held after evict: %q %v", token, err) } } -- 2.47.3