From 18c58dc9c7abb5926a715e3a3bf44ae11ebf048e Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 18 Oct 2026 02:07:52 +1100 Subject: [PATCH 1/2] Add gems virtual repositories (#154) Bundler needs one source for internal and public gems, but gems virtuals have no merger, so a local gems repo and rubygems.org cannot be served together. - merge `info/` across members (member order wins a duplicate version-platform); pass a single-member gem through verbatim - regenerate `versions` lines and md5s for gems several members list; union `names` - 302 `gems/*.gem` and `quick/Marshal.4.8/*` to the member owning that version, via the shared member-path validation - return 502 on a member error; cache merges in the shared virtual cache; route HEAD on virtuals - add `docs/gems/virtual.md` and a stock `ruby:3.3` bundler e2e-docker case Reviewed-on: https://git.unkin.net/unkin/artifactapi/pulls/154 Co-authored-by: unkin-agent Co-committed-by: unkin-agent --- README.md | 2 +- docs/gems/virtual.md | 28 +++ e2e-docker/README.md | 4 +- e2e-docker/gems_virtual_test.go | 69 +++++++ internal/api/v1/gems_virtual.go | 39 ++++ internal/api/v1/proxy.go | 6 + internal/server/server.go | 1 + internal/virtual/engine.go | 12 ++ internal/virtual/gems_merger.go | 294 +++++++++++++++++++++++++++ internal/virtual/gems_merger_test.go | 180 ++++++++++++++++ 10 files changed, 633 insertions(+), 2 deletions(-) create mode 100644 docs/gems/virtual.md create mode 100644 e2e-docker/gems_virtual_test.go create mode 100644 internal/api/v1/gems_virtual.go create mode 100644 internal/virtual/gems_merger.go create mode 100644 internal/virtual/gems_merger_test.go diff --git a/README.md b/README.md index 88a96e4..bfe1eb9 100644 --- a/README.md +++ b/README.md @@ -31,7 +31,7 @@ Each type's paths, Terraform examples and local/virtual behaviour are in `docs/< - `deb`: [remote](docs/deb/remote.md), [remote-github](docs/deb/remote-github.md) - `docker`: [remote](docs/docker/remote.md), [local](docs/docker/local.md) - `generic`: [remote](docs/generic/remote.md) -- `gems`: [remote](docs/gems/remote.md), [local](docs/gems/local.md) +- `gems`: [remote](docs/gems/remote.md), [local](docs/gems/local.md), [virtual](docs/gems/virtual.md) - `goproxy`: [remote](docs/goproxy/remote.md) - `helm`: [remote](docs/helm/remote.md), [local](docs/helm/local.md), [virtual](docs/helm/virtual.md) - `npm`: [remote](docs/npm/remote.md), [local](docs/npm/local.md), [virtual](docs/npm/virtual.md) diff --git a/docs/gems/virtual.md b/docs/gems/virtual.md new file mode 100644 index 0000000..e894113 --- /dev/null +++ b/docs/gems/virtual.md @@ -0,0 +1,28 @@ +# `gems` virtual + +A gems virtual serves one RubyGems source over local and remote gems members: +`gem install` and `bundle install` work against `/api/v1/virtual//`. + +- `info/` is the union of member lines; on a duplicate version-platform + the earlier member wins. A gem only one member has is served verbatim. +- `versions` regenerates the line (and md5) of every gem several members list + from its merged `info`; other gems' lines pass through. `names` is the union. +- `gems/*.gem` and `quick/Marshal.4.8/*.gemspec.rz` 302 to the member whose + `info` line won. +- A member error fails the request with 502; a member without the gem is + skipped. Merges are cached for 60s. + +```hcl +resource "artifactapi_virtual" "gems" { + name = "gems" + package_type = "gems" + members = [artifactapi_local_gems.internal.name, artifactapi_remote_gems.rubygems.name] +} +``` + +```ruby +# Gemfile +source "https://artifactapi.example.com/api/v1/virtual/gems/" +``` + +Push to the local member directly; the virtual is read-only. diff --git a/e2e-docker/README.md b/e2e-docker/README.md index f2c9092..cd51e63 100644 --- a/e2e-docker/README.md +++ b/e2e-docker/README.md @@ -33,7 +33,9 @@ already-running stack. `gem install`, `bundle install` + `require`, republish rejected, `gem yank`). - **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). + client (merged versions, install through the member redirect); gems compact + index merge over two locals via stock `ruby:3.3` bundler (member order wins, + downloads 302 to the owning member). - **Mirrorlist** — an rpm remote with a `mirrorlist` of extra upstream mirrors (pool = `base_url` + `mirrorlist`): round-robin distribution across both mirrors (constant-body `mockupstreama` / `mockupstreamb`), failover past a dead primary, diff --git a/e2e-docker/gems_virtual_test.go b/e2e-docker/gems_virtual_test.go new file mode 100644 index 0000000..1eae0be --- /dev/null +++ b/e2e-docker/gems_virtual_test.go @@ -0,0 +1,69 @@ +//go:build dockere2e + +package e2edocker + +import ( + "fmt" + "os" + "os/exec" + "strings" + "testing" + "time" +) + +// TestVirtualGemsInstall pushes to two local gems repos and drives stock +// bundler through a virtual over both: merged versions, member order winning a +// duplicate version, and downloads redirected to the owning member. +func TestVirtualGemsInstall(t *testing.T) { + network := os.Getenv("COMPOSE_NETWORK") + internal := os.Getenv("ARTIFACTAPI_INTERNAL") + if network == "" || internal == "" { + t.Skip("COMPOSE_NETWORK/ARTIFACTAPI_INTERNAL not set; run via scripts/docker-e2e.sh") + } + if _, err := exec.LookPath("docker"); err != nil { + t.Skip("docker not available on the test host") + } + + id := time.Now().UnixNano() + a, b, virt := fmt.Sprintf("e2e-gems-a-%d", id), fmt.Sprintf("e2e-gems-b-%d", id), fmt.Sprintf("e2e-gems-v-%d", id) + for _, name := range []string{a, b} { + createRepo(t, fmt.Sprintf(`{"name":%q,"package_type":"gems","repo_type":"local"}`, name)) + defer deleteRepo(t, name) + } + createVirtual(t, fmt.Sprintf(`{"name":%q,"package_type":"gems","members":[%q,%q]}`, virt, a, b)) + defer deleteVirtual(t, virt) + + base := strings.TrimRight(internal, "/") + script := `set -euo pipefail +export GEM_HOST_API_KEY=unused +pub() { d=/w/src-$1-$2; mkdir -p $d/lib; cd $d + echo "module UnkinE2eHello; VERSION = '$1:$2'; end" > 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"] }' "$2" > g.gemspec + gem build g.gemspec >/dev/null 2>&1; gem push --host "$BASE/api/v1/local/$1" unkin-e2e-hello-$2.gem; } +pub "$A" 0.1.0 +pub "$B" 0.1.0 +pub "$B" 0.2.0 +V="$BASE/api/v1/virtual/$V" +echo "VERSIONS $(curl -s "$V/versions" | grep '^unkin-e2e-hello ' | cut -d' ' -f2)" +mkdir -p /w/app && cd /w/app +printf 'source "%s/"\ngem "unkin-e2e-hello", "0.1.0"\n' "$V" > Gemfile +bundle install +bundle exec ruby -e 'require "unkin-e2e-hello"; puts "REQ #{UnkinE2eHello::VERSION}"' +echo "REDIRECT $(curl -s -o /dev/null -w '%{redirect_url}' "$V/gems/unkin-e2e-hello-0.2.0.gem")"` + + out, err := exec.Command("docker", "run", "--rm", "--network", network, + "-e", "BASE="+base, "-e", "A="+a, "-e", "B="+b, "-e", "V="+virt, + "ruby:3.3", "bash", "-c", script).CombinedOutput() + if err != nil { + t.Fatalf("bundler run failed: %v\n%s", err, out) + } + for _, want := range []string{ + "VERSIONS 0.1.0,0.2.0", + "REQ " + a + ":0.1.0", + "REDIRECT " + base + "/api/v1/local/" + b + "/gems/unkin-e2e-hello-0.2.0.gem", + } { + if !strings.Contains(string(out), want) { + t.Fatalf("missing %q in bundler output:\n%s", want, out) + } + } +} diff --git a/internal/api/v1/gems_virtual.go b/internal/api/v1/gems_virtual.go new file mode 100644 index 0000000..fbaf751 --- /dev/null +++ b/internal/api/v1/gems_virtual.go @@ -0,0 +1,39 @@ +package v1 + +import ( + "errors" + "log/slog" + "net/http" + + "git.unkin.net/unkin/artifactapi/internal/virtual" + "git.unkin.net/unkin/artifactapi/pkg/models" +) + +// serveVirtualGems serves the merged compact index and 302s downloads to the +// member owning that version. +func (h *ProxyHandler) serveVirtualGems(w http.ResponseWriter, r *http.Request, virt models.Virtual, path, proxyBaseURL string) { + var body []byte + var loc string + var err error + if isGemsIndexPath(path) { + body, _, err = h.virtualEngine.FetchGems(r.Context(), virt, path) + } else { + loc, err = h.virtualEngine.GemsRedirect(r.Context(), virt, path, r.URL.RawQuery, proxyBaseURL) + } + switch { + case errors.Is(err, virtual.ErrNotFound): + gemsText(w, http.StatusNotFound, "not found") + case errors.Is(err, virtual.ErrBadPath): + gemsText(w, http.StatusBadRequest, err.Error()) + case err != nil: + slog.Error("virtual gems fetch failed", "virtual", virt.Name, "path", path, "error", err) + gemsText(w, http.StatusBadGateway, "bad gateway") + case loc != "": + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("Vary", "Host, X-Forwarded-Proto") + http.Redirect(w, r, loc, http.StatusFound) + default: + w.Header().Set("X-Artifact-Source", "virtual") + gemsText(w, http.StatusOK, string(body)) + } +} diff --git a/internal/api/v1/proxy.go b/internal/api/v1/proxy.go index 16ebc55..c361142 100644 --- a/internal/api/v1/proxy.go +++ b/internal/api/v1/proxy.go @@ -52,6 +52,7 @@ func (h *ProxyHandler) Routes() chi.Router { r.Post("/local/{localName}/*", h.handleLocalPost) r.Delete("/local/{localName}/*", h.handleLocalDelete) r.Get("/virtual/{virtualName}/*", h.handleVirtual) + r.Head("/virtual/{virtualName}/*", h.handleVirtual) return r } @@ -204,6 +205,11 @@ func (h *ProxyHandler) handleVirtual(w http.ResponseWriter, r *http.Request) { return } + if virt.PackageType == models.PackageGems { + h.serveVirtualGems(w, r, *virt, path, proxyBaseURL) + return + } + loc, ok, err := h.virtualEngine.MemberRedirect(r.Context(), *virt, path, r.URL.RawQuery, proxyBaseURL) if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) diff --git a/internal/server/server.go b/internal/server/server.go index b9d8a35..26a4953 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -167,6 +167,7 @@ func (s *Server) routes() chi.Router { proxyHandler := v1.NewProxyHandler(s.engine, s.virtEngine, s.db, s.store, s.localHandler) s.virtEngine.NPMLocal = proxyHandler.LocalNPMPackument + s.virtEngine.GemsLocal = proxyHandler.LocalGemsIndex r.Mount("/api/v1", proxyHandler.Routes()) r.Mount("/v2", proxyHandler.DockerV2Routes()) diff --git a/internal/virtual/engine.go b/internal/virtual/engine.go index 0ac8687..99457e2 100644 --- a/internal/virtual/engine.go +++ b/internal/virtual/engine.go @@ -32,6 +32,12 @@ type Engine struct { // to the repo, or ErrNotFound. NPMLocal func(ctx context.Context, repo, pkg string) ([]byte, error) + gemsMember func(ctx context.Context, name, path string) (*GemsMember, error) + + // GemsLocal returns a local gems repo's versions, names or info/, or + // ErrNotFound. + GemsLocal func(ctx context.Context, repo, path string) ([]byte, error) + mergeTTL time.Duration sf singleflight.Group @@ -61,6 +67,7 @@ func NewEngine(db *database.DB, proxyEngine *proxy.Engine) *Engine { e := &Engine{db: db, proxyEngine: proxyEngine, getRemote: db.GetRemote, mergeTTL: rpmMergeTTL} e.rpmMember = e.fetchRPMMember e.npmMember = e.fetchNPMMember + e.gemsMember = e.fetchGemsMember return e } @@ -202,6 +209,11 @@ func (e *Engine) MemberRedirect(ctx context.Context, virt models.Virtual, path, if !found || !slices.Contains(virt.Members, name) { return "", false, nil } + return e.memberLocation(ctx, name, rest, rawQuery, proxyBaseURL) +} + +// memberLocation returns the URL of rest on member name's own route. +func (e *Engine) memberLocation(ctx context.Context, name, rest, rawQuery, proxyBaseURL string) (string, bool, error) { segs, err := memberSegments(rest) if err != nil { return "", false, err diff --git a/internal/virtual/gems_merger.go b/internal/virtual/gems_merger.go new file mode 100644 index 0000000..4fdebaf --- /dev/null +++ b/internal/virtual/gems_merger.go @@ -0,0 +1,294 @@ +package virtual + +import ( + "bytes" + "context" + "errors" + "fmt" + "net/http" + "slices" + "strings" + "time" + + "git.unkin.net/unkin/artifactapi/internal/provider/gems" + "git.unkin.net/unkin/artifactapi/internal/proxy" + "git.unkin.net/unkin/artifactapi/pkg/models" +) + +const ( + maxGemsInfos = 1024 + maxGemsCandidates = 8 +) + +// GemsMember is one member's compact index document; Body is nil when the +// member does not have it. +type GemsMember struct { + RemoteName string + Body []byte +} + +// GemsDoc is a merged compact index document. Owners maps each /info +// version[-platform] to the member serving it. +type GemsDoc struct { + Body []byte + Owners map[string]string +} + +func (d *GemsDoc) index() []byte { return d.Body } +func (d *GemsDoc) files() map[string][]byte { return nil } + +// MergeGemsInfo unions /info lines across members; the first member wins a +// duplicate version-platform. A gem only one member has is passed through +// verbatim so its md5 matches that member's /versions. ErrNotFound means no +// member has the gem. +func MergeGemsInfo(members []GemsMember) (*GemsDoc, error) { + var have []GemsMember + for _, m := range members { + if m.Body != nil { + have = append(have, m) + } + } + if len(have) == 0 { + return nil, ErrNotFound + } + d := &GemsDoc{Owners: map[string]string{}} + var lines []string + for _, m := range have { + for _, l := range gems.InfoLines(m.Body) { + k := gems.InfoKey(l) + if _, dup := d.Owners[k]; dup { + continue + } + d.Owners[k] = m.RemoteName + lines = append(lines, l) + } + } + if len(have) == 1 { + d.Body = have[0].Body + } else { + d.Body = gems.InfoBody(lines) + } + return d, nil +} + +// versionsLines returns the entry lines of a /versions document. +func versionsLines(m GemsMember) ([]byte, error) { + if m.Body == nil { + return nil, fmt.Errorf("member %q has no versions", m.RemoteName) + } + if bytes.HasPrefix(m.Body, []byte("---\n")) { + return m.Body[4:], nil + } + _, rest, ok := bytes.Cut(m.Body, []byte("\n---\n")) + if !ok { + return nil, fmt.Errorf("member %q versions has no --- separator", m.RemoteName) + } + return rest, nil +} + +func gemName(line []byte) string { + n, _, _ := bytes.Cut(line, []byte(" ")) + return string(bytes.TrimSpace(n)) +} + +// MergeGemsVersions passes through the lines of gems only one member lists +// and writes one regenerated line, from the merged /info, per gem that +// several members list. +func MergeGemsVersions(members []GemsMember, info func(name string) (*GemsDoc, error), at time.Time) ([]byte, error) { + bodies := make([][]byte, len(members)) + owners := map[string]int{} + for i, m := range members { + rest, err := versionsLines(m) + if err != nil { + return nil, err + } + bodies[i] = rest + seen := map[string]bool{} + for line := range bytes.Lines(rest) { + if n := gemName(line); n != "" && !seen[n] { + seen[n] = true + owners[n]++ + } + } + } + var b bytes.Buffer + b.WriteString(gems.VersionsHeader(at)) + var shared []string + for _, rest := range bodies { + for line := range bytes.Lines(rest) { + n := gemName(line) + if n == "" { + continue + } + if owners[n] > 1 { + if !slices.Contains(shared, n) { + shared = append(shared, n) + } + continue + } + b.Write(bytes.TrimRight(line, "\n")) + b.WriteByte('\n') + } + } + for _, n := range shared { + d, err := info(n) + if errors.Is(err, ErrNotFound) { + continue + } + if err != nil { + return nil, err + } + b.WriteString(gems.VersionsLine(n, d.Body) + "\n") + } + return b.Bytes(), nil +} + +// MergeGemsNames unions member /names documents, sorted. +func MergeGemsNames(members []GemsMember) ([]byte, error) { + set := map[string]bool{} + for _, m := range members { + rest, err := versionsLines(m) + if err != nil { + return nil, err + } + for line := range bytes.Lines(rest) { + if n := gemName(line); n != "" { + set[n] = true + } + } + } + names := make([]string, 0, len(set)) + for n := range set { + names = append(names, n) + } + slices.Sort(names) + return []byte("---\n" + strings.Join(names, "\n") + "\n"), nil +} + +// FetchGems serves a gems virtual's merged versions, names or info/. +func (e *Engine) FetchGems(ctx context.Context, virt models.Virtual, path string) ([]byte, string, error) { + var d *GemsDoc + var err error + switch { + case strings.HasPrefix(path, "info/"): + name := strings.TrimPrefix(path, "info/") + if !gems.NameRe.MatchString(name) { + return nil, "", ErrNotFound + } + d, err = e.gemsInfo(ctx, virt, name) + case path == "versions": + d, err = e.gemsMerged(ctx, virt, path, func(ctx context.Context, ms []GemsMember) ([]byte, error) { + return MergeGemsVersions(ms, func(name string) (*GemsDoc, error) { return e.gemsInfo(ctx, virt, name) }, time.Now()) + }) + case path == "names": + d, err = e.gemsMerged(ctx, virt, path, func(_ context.Context, ms []GemsMember) ([]byte, error) { + return MergeGemsNames(ms) + }) + default: + return nil, "", ErrNotFound + } + if err != nil { + return nil, "", err + } + return d.Body, "text/plain; charset=utf-8", nil +} + +func (e *Engine) gemsMerged(ctx context.Context, virt models.Virtual, path string, merge func(context.Context, []GemsMember) ([]byte, error)) (*GemsDoc, error) { + repo, err := e.merged(ctx, "gems\x00"+virt.Name+"\x00"+path, func(ctx context.Context) (mergedRepo, error) { + members, err := fetchMembers(virt, func(name string) (*GemsMember, error) { return e.gemsMember(ctx, name, path) }) + if err != nil { + return nil, err + } + body, err := merge(ctx, members) + if err != nil { + return nil, err + } + return &GemsDoc{Body: body}, nil + }) + if err != nil { + return nil, err + } + return repo.(*GemsDoc), nil +} + +func (e *Engine) gemsInfo(ctx context.Context, virt models.Virtual, name string) (*GemsDoc, error) { + prefix := "gems\x00" + virt.Name + "\x00info/" + inserted := false + repo, err := e.merged(ctx, prefix+name, func(ctx context.Context) (mergedRepo, error) { + inserted = true + members, err := fetchMembers(virt, func(m string) (*GemsMember, error) { return e.gemsMember(ctx, m, "info/"+name) }) + if err != nil { + return nil, err + } + return MergeGemsInfo(members) + }) + if err != nil { + return nil, err + } + if inserted { + e.evictOldest(prefix, maxGemsInfos) + } + return repo.(*GemsDoc), nil +} + +// GemsRedirect maps a gems virtual download (gems/.gem or +// quick/Marshal.4.8/.gemspec.rz) to the member whose /info line won the +// merge for that version-platform. +func (e *Engine) GemsRedirect(ctx context.Context, virt models.Virtual, path, rawQuery, proxyBaseURL string) (string, error) { + full, ok := strings.CutPrefix(path, "gems/") + if ok { + full, ok = strings.CutSuffix(full, ".gem") + } else if full, ok = strings.CutPrefix(path, "quick/Marshal.4.8/"); ok { + full, ok = strings.CutSuffix(full, ".gemspec.rz") + } + if !ok || strings.Contains(full, "/") { + return "", ErrNotFound + } + tried := 0 + for i := 1; i < len(full)-1 && tried < maxGemsCandidates; i++ { + if full[i] != '-' || full[i+1] < '0' || full[i+1] > '9' || !gems.NameRe.MatchString(full[:i]) { + continue + } + tried++ + d, err := e.gemsInfo(ctx, virt, full[:i]) + if errors.Is(err, ErrNotFound) { + continue + } + if err != nil { + return "", err + } + if owner, ok := d.Owners[full[i+1:]]; ok { + loc, ok, err := e.memberLocation(ctx, owner, path, rawQuery, proxyBaseURL) + if err == nil && !ok { + err = ErrNotFound + } + return loc, err + } + } + return "", ErrNotFound +} + +// fetchGemsMember returns a nil Body when the member lacks the document. +func (e *Engine) fetchGemsMember(ctx context.Context, name, path string) (*GemsMember, error) { + remote, err := e.getRemote(ctx, name) + if err != nil { + return nil, fmt.Errorf("remote %q: %w", name, err) + } + m := &GemsMember{RemoteName: name} + if remote.RepoType == models.RepoTypeLocal { + if e.GemsLocal == nil { + return nil, errors.New("local gems indexes are not wired") + } + m.Body, err = e.GemsLocal(ctx, name, path) + } else { + m.Body, err = e.fetchMemberPath(ctx, *remote, path) + } + var pe *proxy.ProxyError + if errors.Is(err, ErrNotFound) || errors.As(err, &pe) && pe.Status == http.StatusNotFound { + return &GemsMember{RemoteName: name}, nil + } + if err != nil { + return nil, err + } + return m, nil +} diff --git a/internal/virtual/gems_merger_test.go b/internal/virtual/gems_merger_test.go new file mode 100644 index 0000000..5b2a4bd --- /dev/null +++ b/internal/virtual/gems_merger_test.go @@ -0,0 +1,180 @@ +package virtual + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + "git.unkin.net/unkin/artifactapi/internal/provider/gems" + "git.unkin.net/unkin/artifactapi/pkg/models" +) + +func TestMergeGemsInfoMemberOrderWins(t *testing.T) { + local := GemsMember{RemoteName: "internal", Body: []byte("---\n1.0.0 |checksum:aa\n1.0.0-java |checksum:ab\n")} + remote := GemsMember{RemoteName: "rubygems", Body: []byte("---\n0.9.0 |checksum:99\n1.0.0 |checksum:bb\n2.0.0 rack:>= 0|checksum:cc,ruby:>= 3.0\n")} + d, err := MergeGemsInfo([]GemsMember{local, {RemoteName: "empty"}, remote}) + if err != nil { + t.Fatal(err) + } + want := "---\n1.0.0 |checksum:aa\n1.0.0-java |checksum:ab\n0.9.0 |checksum:99\n2.0.0 rack:>= 0|checksum:cc,ruby:>= 3.0\n" + if string(d.Body) != want { + t.Fatalf("merged info =\n%q\nwant\n%q", d.Body, want) + } + for k, owner := range map[string]string{"1.0.0": "internal", "1.0.0-java": "internal", "0.9.0": "rubygems", "2.0.0": "rubygems"} { + if d.Owners[k] != owner { + t.Errorf("owner of %s = %q, want %q", k, d.Owners[k], owner) + } + } + + odd := GemsMember{RemoteName: "rubygems", Body: []byte("---\n1.0.0 |checksum:bb\n\n")} + if d, _ := MergeGemsInfo([]GemsMember{{RemoteName: "internal"}, odd}); string(d.Body) != string(odd.Body) { + t.Fatalf("a gem only one member has must pass through verbatim, got %q", d.Body) + } + if _, err := MergeGemsInfo([]GemsMember{{RemoteName: "a"}}); !errors.Is(err, ErrNotFound) { + t.Fatalf("want ErrNotFound, got %v", err) + } +} + +func TestMergeGemsVersions(t *testing.T) { + local := GemsMember{RemoteName: "internal", Body: []byte("created_at: 2026-10-18T00:00:00Z\n---\nhello 0.1.0 11\nrack 9.0.0 22\n")} + remote := GemsMember{RemoteName: "rubygems", Body: []byte("created_at: 2024-04-01T00:00:05Z\n---\n" + + "rack 1.0.0,2.0.0 33\nrails 7.0.0 44\nrack -1.0.0,3.0.0 55\n")} + merged := &GemsDoc{Body: []byte("---\n9.0.0 |checksum:x\n2.0.0 |checksum:y\n3.0.0 |checksum:z\n")} + var asked []string + at := time.Date(2026, 10, 18, 1, 2, 3, 0, time.UTC) + body, err := MergeGemsVersions([]GemsMember{local, remote}, func(name string) (*GemsDoc, error) { + asked = append(asked, name) + return merged, nil + }, at) + if err != nil { + t.Fatal(err) + } + want := "created_at: 2026-10-18T01:02:03Z\n---\nhello 0.1.0 11\nrails 7.0.0 44\nrack 9.0.0,2.0.0,3.0.0 " + gems.MD5Hex(merged.Body) + "\n" + if string(body) != want { + t.Fatalf("versions =\n%q\nwant\n%q", body, want) + } + if len(asked) != 1 || asked[0] != "rack" { + t.Fatalf("merged info fetched for %v, want only the shared gem", asked) + } + + if _, err := MergeGemsVersions([]GemsMember{{RemoteName: "x", Body: []byte("no separator")}}, nil, at); err == nil { + t.Fatal("malformed member versions must fail") + } + if _, err := MergeGemsVersions([]GemsMember{{RemoteName: "x"}}, nil, at); err == nil { + t.Fatal("a member without versions must fail") + } +} + +func TestMergeGemsNames(t *testing.T) { + body, err := MergeGemsNames([]GemsMember{{RemoteName: "a", Body: []byte("---\nzeta\nalpha\n")}, {RemoteName: "b", Body: []byte("---\nalpha\nmid\n")}}) + if err != nil || string(body) != "---\nalpha\nmid\nzeta\n" { + t.Fatalf("names = %q %v", body, err) + } +} + +func gemsEngine(fetch func(name, path string) (*GemsMember, error)) *Engine { + return &Engine{ + mergeTTL: time.Minute, + getRemote: func(_ context.Context, name string) (*models.Remote, error) { + rt := models.RepoTypeRemote + if name == "internal" { + rt = models.RepoTypeLocal + } + return &models.Remote{Name: name, RepoType: rt}, nil + }, + gemsMember: func(_ context.Context, name, path string) (*GemsMember, error) { return fetch(name, path) }, + } +} + +func gemsVirt(members ...string) models.Virtual { + return models.Virtual{Name: "gems", PackageType: models.PackageGems, Members: members} +} + +func TestGemsRedirectToOwner(t *testing.T) { + var mu sync.Mutex + asked := map[string]int{} + e := gemsEngine(func(name, path string) (*GemsMember, error) { + mu.Lock() + asked[name+" "+path]++ + mu.Unlock() + switch name + " " + path { + case "internal info/foo-2fa": + return &GemsMember{RemoteName: name, Body: []byte("---\n1.0.0 |checksum:a\n")}, nil + case "rubygems info/foo-2fa": + return &GemsMember{RemoteName: name, Body: []byte("---\n1.0.0 |checksum:b\n1.1.0-x86_64-linux |checksum:c\n")}, nil + case "rubygems info/foo": + return &GemsMember{RemoteName: name, Body: []byte("---\n2.0.0 |checksum:d\n")}, nil + } + return &GemsMember{RemoteName: name}, nil + }) + virt := gemsVirt("internal", "rubygems") + for path, want := range map[string]string{ + "gems/foo-2fa-1.0.0.gem": "http://host/api/v1/local/internal/gems/foo-2fa-1.0.0.gem", + "gems/foo-2fa-1.1.0-x86_64-linux.gem": "http://host/api/v1/remote/rubygems/gems/foo-2fa-1.1.0-x86_64-linux.gem", + "quick/Marshal.4.8/foo-2fa-1.0.0.gemspec.rz": "http://host/api/v1/local/internal/quick/Marshal.4.8/foo-2fa-1.0.0.gemspec.rz", + "quick/Marshal.4.8/foo-2.0.0.gemspec.rz": "http://host/api/v1/remote/rubygems/quick/Marshal.4.8/foo-2.0.0.gemspec.rz", + } { + loc, err := e.GemsRedirect(context.Background(), virt, path, "", "http://host") + if err != nil || loc != want { + t.Errorf("%s -> %q %v, want %q", path, loc, err, want) + } + } + for _, path := range []string{"gems/foo-2fa-9.9.9.gem", "gems/nodash.gem", "gems/x/../foo-2.0.0.gem", "other/foo-2.0.0.gem", "specs.4.8.gz"} { + if _, err := e.GemsRedirect(context.Background(), virt, path, "", "http://host"); !errors.Is(err, ErrNotFound) { + t.Errorf("%s: want ErrNotFound, got %v", path, err) + } + } + if asked["internal info/foo-2fa"] != 1 { + t.Fatalf("merged info must be cached across redirects, internal asked %d times", asked["internal info/foo-2fa"]) + } +} + +func TestFetchGemsFailsClosed(t *testing.T) { + e := gemsEngine(func(name, path string) (*GemsMember, error) { + if name == "rubygems" { + return nil, errors.New("upstream down") + } + return &GemsMember{RemoteName: name, Body: []byte("created_at: x\n---\nhello 0.1.0 11\n")}, nil + }) + for _, path := range []string{"versions", "names", "info/hello"} { + if _, _, err := e.FetchGems(context.Background(), gemsVirt("internal", "rubygems"), path); err == nil || errors.Is(err, ErrNotFound) { + t.Errorf("%s: want member error, got %v", path, err) + } + } + if _, err := e.GemsRedirect(context.Background(), gemsVirt("internal", "rubygems"), "gems/hello-0.1.0.gem", "", "http://h"); err == nil || errors.Is(err, ErrNotFound) { + t.Errorf("redirect: want member error, got %v", err) + } +} + +func TestFetchGemsVersionsUsesMergedInfo(t *testing.T) { + e := gemsEngine(func(name, path string) (*GemsMember, error) { + docs := map[string]string{ + "internal versions": "created_at: x\n---\nrack 9.0.0 11\n", + "rubygems versions": "created_at: y\n---\nrack 3.0.0 22\nrails 7.0.0 33\n", + "internal info/rack": "---\n9.0.0 |checksum:a\n", + "rubygems info/rack": "---\n3.0.0 |checksum:b\n", + } + if d, ok := docs[name+" "+path]; ok { + return &GemsMember{RemoteName: name, Body: []byte(d)}, nil + } + return &GemsMember{RemoteName: name}, nil + }) + virt := gemsVirt("internal", "rubygems") + info, ct, err := e.FetchGems(context.Background(), virt, "info/rack") + if err != nil || !strings.HasPrefix(ct, "text/plain") { + t.Fatalf("info: %v %q", err, ct) + } + versions, _, err := e.FetchGems(context.Background(), virt, "versions") + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(versions), "\nrails 7.0.0 33\nrack 9.0.0,3.0.0 "+gems.MD5Hex(info)+"\n") { + t.Fatalf("versions = %q", versions) + } + if _, _, err := e.FetchGems(context.Background(), virt, "info/../x"); !errors.Is(err, ErrNotFound) { + t.Fatalf("bad name: %v", err) + } +} From f9791d4b089371aa74bf507ab683828cebe5a0d3 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 18 Oct 2026 02:08:46 +1100 Subject: [PATCH 2/2] 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//` 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: https://git.unkin.net/unkin/artifactapi/pulls/155 Co-authored-by: unkin-agent Co-committed-by: unkin-agent --- e2e-docker/README.md | 3 +- e2e-docker/gems_local_test.go | 15 +++-- internal/api/v1/gems.go | 85 ++++++++++++++++++++---- internal/api/v1/gems_handler_test.go | 98 +++++++++++++++++++++++++++- internal/api/v1/npm_handler_test.go | 18 ++++- internal/api/v1/proxy.go | 12 +++- internal/provider/gems/local.go | 7 +- internal/provider/gems/local_test.go | 9 +++ internal/proxy/engine.go | 24 ++++--- 9 files changed, 234 insertions(+), 37 deletions(-) diff --git a/e2e-docker/README.md b/e2e-docker/README.md index cd51e63..4cc17ef 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); gems compact 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..912777f 100644 --- a/internal/api/v1/gems.go +++ b/internal/api/v1/gems.go @@ -26,10 +26,17 @@ 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 +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 } diff --git a/internal/api/v1/gems_handler_test.go b/internal/api/v1/gems_handler_test.go index 095db07..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" ) @@ -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) + } +} 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/api/v1/proxy.go b/internal/api/v1/proxy.go index c361142..dbe083e 100644 --- a/internal/api/v1/proxy.go +++ b/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) 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) { 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 }