Compare commits

..

5 Commits

Author SHA1 Message Date
unkin-agent d88a047788 Fail GitHub scans on asset errors and drop sync state with its remote (#135)
ci/woodpecker/tag/docker Pipeline was successful
A GitHub release scan that hit a transient asset error still saved the release-list ETag, so the failed asset was never retried. Assets that are corrupt, deleted or truncated upstream fail on every retry, so they must not keep the remote failed.

- fail the rpm/deb/alpine scan on transient asset fetch, short read or metadata insert errors, so the ETag is not saved and the backoff retry re-derives the asset
- add `provider.ErrInvalidPackage` for unparseable packages and 404/410/416 asset downloads; such assets are logged and skipped
- decide a ranged fetch holds the whole object from Content-Length or the Content-Range total, or a clean EOF when the total is unknown
- add a NOT VALID cascading FK from the GitHub sync-state tables to `remotes`

Reviewed-on: #135
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-18 02:33:00 +11:00
unkin-agent b0d60b3e8a Release Redis locks only while the caller still owns them (#156)
AcquireLock stored a constant value and ReleaseLock was an unconditional DEL, so a holder that outlived its TTL deleted the next holder's lock. This hits Fetch, Evict and gems push/yank.

- store a random owner token with SET NX PX and return it from AcquireLock
- release via a Lua compare-and-delete keyed on that token
- carry the token through Fetch, Evict and Engine.Lock's release func
- release the Fetch lock with an uncancelled context
- make Engine.Lock's busy 503 message generic

Reviewed-on: #156
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-18 02:32:40 +11:00
unkin-agent f9791d4b08 Keep a tombstone on gem yank (#155)
ci/woodpecker/tag/docker Pipeline was successful
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>
2026-10-18 02:08:46 +11:00
unkin-agent 18c58dc9c7 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/<gem>` 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: #154
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-18 02:07:52 +11:00
unkin-agent 7dd7fc70d1 Add local gems repositories (#153)
In-house gems have nowhere to publish: gems locals only take raw file uploads, which `gem push`, `gem install` and bundler cannot use.

- accept `gem push` (`POST <repo>/api/v1/gems`, 256MiB cap, 8MiB metadata cap) and `gem yank` (`DELETE <repo>/api/v1/gems/yank`); any API key works
- parse the gemspec from the `.gem`; reject an existing name-version-platform with 409
- serve the compact index (`versions`, `info/<gem>`, `names`), `.gem` files and generated `quick/Marshal.4.8` specs; no legacy `specs.4.8.gz`
- route HEAD on locals (gem probes `versions` with HEAD)
- add `docs/gems/local.md`, a UI snippet and a stock `ruby:3.3` e2e-docker case

Reviewed-on: #153
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>
2026-10-18 01:12:37 +11:00
35 changed files with 2626 additions and 94 deletions
+2 -1
View File
@@ -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) - `deb`: [remote](docs/deb/remote.md), [remote-github](docs/deb/remote-github.md)
- `docker`: [remote](docs/docker/remote.md), [local](docs/docker/local.md) - `docker`: [remote](docs/docker/remote.md), [local](docs/docker/local.md)
- `generic`: [remote](docs/generic/remote.md) - `generic`: [remote](docs/generic/remote.md)
- `gems`: [remote](docs/gems/remote.md) - `gems`: [remote](docs/gems/remote.md), [local](docs/gems/local.md), [virtual](docs/gems/virtual.md)
- `goproxy`: [remote](docs/goproxy/remote.md) - `goproxy`: [remote](docs/goproxy/remote.md)
- `helm`: [remote](docs/helm/remote.md), [local](docs/helm/local.md), [virtual](docs/helm/virtual.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) - `npm`: [remote](docs/npm/remote.md), [local](docs/npm/local.md), [virtual](docs/npm/virtual.md)
@@ -131,6 +131,7 @@ No patterns + no blocklist = open proxy. Provider handles mutability classificat
GET /api/v1/remote/{name}/{path} Proxy/cache artifact GET /api/v1/remote/{name}/{path} Proxy/cache artifact
GET /api/v1/virtual/{name}/{path} Virtual repo (merged index) GET /api/v1/virtual/{name}/{path} Virtual repo (merged index)
PUT /api/v1/local/{name}/{package} npm publish (local npm repos) PUT /api/v1/local/{name}/{package} npm publish (local npm repos)
POST/DELETE /api/v1/local/{name}/api/v1/gems[/yank] gem push / yank (local gems repos)
GET /v2/{name}/{path} Docker Registry v2 GET /v2/{name}/{path} Docker Registry v2
``` ```
+24
View File
@@ -0,0 +1,24 @@
# Local gems repository
A local `gems` repo is a RubyGems source: `gem push`, `gem yank`, `gem install`
and `bundle install` work against `/api/v1/local/<repo>`. `gem push` needs an
API key; any value works.
```bash
GEM_HOST_API_KEY=unused gem push --host https://artifactapi.example.com/api/v1/local/<repo> mygem-1.0.0.gem
gem install mygem --source https://artifactapi.example.com/api/v1/local/<repo>/
gem yank mygem -v 1.0.0 --host https://artifactapi.example.com/api/v1/local/<repo>
```
```ruby
# Gemfile
source "https://artifactapi.example.com/api/v1/local/<repo>/"
gem "mygem"
```
The repo serves the compact index (`versions`, `info/<gem>`, `names`), `gems/*.gem`
and `quick/Marshal.4.8/*.gemspec.rz`. The legacy `specs.4.8.gz` indexes are not
served; current gem and bundler clients use the compact index.
Versions are immutable: pushing an existing name-version-platform is rejected
with 409. A yanked version disappears from the index and its files are removed.
+28
View File
@@ -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/<name>/`.
- `info/<gem>` 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.
+6 -2
View File
@@ -29,10 +29,14 @@ already-running stack.
generic (upload/download), pypi (wheel + generated `simple/` generic (upload/download), pypi (wheel + generated `simple/`
index), rpm (real package + **automatic repodata** generation), npm (stock index), rpm (real package + **automatic repodata** generation), npm (stock
`node:22` client: `npm publish` unscoped + scoped, `npm install` + `require()`, `node:22` client: `npm publish` unscoped + scoped, `npm install` + `require()`,
second version, republish rejected). second version, republish rejected), gems (stock `ruby:3.3`: `gem push`,
`gem install`, `bundle install` + `require`, republish rejected, `gem yank`,
yanked re-push rejected).
- **Virtual repositories** — pypi simple-index merge and helm `index.yaml` merge - **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` 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 - **Mirrorlist** — an rpm remote with a `mirrorlist` of extra upstream mirrors
(pool = `base_url` + `mirrorlist`): round-robin distribution across both mirrors (pool = `base_url` + `mirrorlist`): round-robin distribution across both mirrors
(constant-body `mockupstreama` / `mockupstreamb`), failover past a dead primary, (constant-body `mockupstreama` / `mockupstreamb`), failover past a dead primary,
+67
View File
@@ -0,0 +1,67 @@
//go:build dockere2e
package e2edocker
import (
"fmt"
"os"
"os/exec"
"strings"
"testing"
"time"
)
// 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 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")
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")
}
name := fmt.Sprintf("e2e-gems-local-%d", time.Now().UnixNano())
createRepo(t, fmt.Sprintf(`{"name":%q,"package_type":"gems","repo_type":"local"}`, name))
defer deleteRepo(t, name)
src := strings.TrimRight(internal, "/") + "/api/v1/local/" + name
script := `set -euo pipefail
export GEM_HOST_API_KEY=unused
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
mk 0.2.0 && gem push --host "$SRC" unkin-e2e-hello-0.2.0.gem
if gem push --host "$SRC" unkin-e2e-hello-0.2.0.gem; then echo "republish accepted"; exit 1; fi
mkdir -p /w/app && cd /w/app
printf -- "---\n:sources:\n- $SRC/\n" > ~/.gemrc
gem install --no-document unkin-e2e-hello
ruby -e 'require "unkin-e2e-hello"; puts "GEM #{UnkinE2eHello::VERSION}"'
gem uninstall -a -x unkin-e2e-hello >/dev/null
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' ' ')"
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 ", "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)
}
}
}
+69
View File
@@ -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)
}
}
}
+308
View File
@@ -0,0 +1,308 @@
package v1
import (
"cmp"
"context"
"errors"
"fmt"
"io"
"log/slog"
"maps"
"net/http"
"net/url"
"slices"
"strings"
"time"
"github.com/go-chi/chi/v5"
"git.unkin.net/unkin/artifactapi/internal/database"
"git.unkin.net/unkin/artifactapi/internal/provider/gems"
"git.unkin.net/unkin/artifactapi/internal/virtual"
"git.unkin.net/unkin/artifactapi/pkg/models"
)
// Local gems repos speak the RubyGems push/yank API and compact index. Per gem:
// 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
gemsLockWait = 30 * time.Second
)
const (
gemsIndexPrefix = "_index/"
gemsYankedPrefix = "_yanked/"
)
func gemsText(w http.ResponseWriter, status int, msg string) {
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
w.WriteHeader(status)
_, _ = io.WriteString(w, msg)
}
func (h *ProxyHandler) localGemsRemote(w http.ResponseWriter, r *http.Request) (*models.Remote, bool) {
remote, err := h.db.GetRemote(r.Context(), chi.URLParam(r, "localName"))
if err != nil || remote.RepoType != models.RepoTypeLocal || remote.PackageType != models.PackageGems {
gemsText(w, http.StatusMethodNotAllowed, "only supported on local gems repositories")
return nil, false
}
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")
return
}
remote, ok := h.localGemsRemote(w, r)
if !ok {
return
}
gem, err := io.ReadAll(http.MaxBytesReader(w, r.Body, gemsMaxPushBytes))
if err != nil {
gemsText(w, http.StatusRequestEntityTooLarge, fmt.Sprintf("gem exceeds %d bytes", gemsMaxPushBytes))
return
}
spec, err := gems.ParseGem(gem)
if err != nil {
gemsText(w, http.StatusBadRequest, err.Error())
return
}
quick, err := gems.QuickSpec(spec)
if err != nil {
gemsText(w, http.StatusInternalServerError, err.Error())
return
}
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
}{
{paths[0], gem},
{paths[1], quick},
} {
ok, err := h.npmStoreImmutable(r.Context(), remote.Name, f.path, f.data, "application/octet-stream")
if err != nil {
fail(http.StatusInternalServerError, err.Error())
return
}
if !ok {
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 {
fail(http.StatusInternalServerError, err.Error())
return
}
slog.Info("local gem pushed", "repo", remote.Name, "gem", full)
gemsText(w, http.StatusOK, "Successfully registered gem: "+spec.Name+" ("+spec.Version+")")
}
func (h *ProxyHandler) handleLocalDelete(w http.ResponseWriter, r *http.Request) {
if chi.URLParam(r, "*") != "api/v1/gems/yank" {
gemsText(w, http.StatusNotFound, "not found")
return
}
remote, ok := h.localGemsRemote(w, r)
if !ok {
return
}
body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 64<<10))
if err != nil {
gemsText(w, http.StatusRequestEntityTooLarge, "yank body too large")
return
}
form, err := url.ParseQuery(string(body))
if err != nil {
gemsText(w, http.StatusBadRequest, "invalid form body")
return
}
for k, v := range r.URL.Query() {
if form.Get(k) == "" {
form[k] = v
}
}
spec := &gems.Spec{Name: form.Get("gem_name"), Version: form.Get("version"), Platform: form.Get("platform")}
if spec.Platform == "" {
spec.Platform = "ruby"
}
if !gems.NameRe.MatchString(spec.Name) || !gems.VersionRe.MatchString(spec.Version) || !gems.PlatformRe.MatchString(spec.Platform) {
gemsText(w, http.StatusBadRequest, "invalid gem_name, version or platform")
return
}
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 {
gemsText(w, http.StatusInternalServerError, err.Error())
return
}
if existing == nil {
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(context.WithoutCancel(r.Context()), remote.Name, p); err != nil {
gemsText(w, http.StatusInternalServerError, err.Error())
return
}
}
slog.Info("local gem yanked", "repo", remote.Name, "gem", full)
gemsText(w, http.StatusOK, "Successfully deleted gem: "+spec.Name+" ("+spec.Version+")")
}
// isGemsIndexPath reports whether path is a compact index document.
func isGemsIndexPath(path string) bool {
return path == "versions" || path == "names" || strings.HasPrefix(path, "info/")
}
func (h *ProxyHandler) serveLocalGemsIndex(w http.ResponseWriter, r *http.Request, remote *models.Remote, path string) {
body, err := h.LocalGemsIndex(r.Context(), remote.Name, path)
if errors.Is(err, virtual.ErrNotFound) {
gemsText(w, http.StatusNotFound, "not found")
return
}
if err != nil {
gemsText(w, http.StatusInternalServerError, err.Error())
return
}
w.Header().Set("X-Artifact-Source", "local")
gemsText(w, http.StatusOK, string(body))
}
// LocalGemsIndex builds a local gems repo's versions, names or info/<gem>
// document, or returns virtual.ErrNotFound.
func (h *ProxyHandler) LocalGemsIndex(ctx context.Context, repo, path string) ([]byte, error) {
if name, ok := strings.CutPrefix(path, "info/"); ok {
if !gems.NameRe.MatchString(name) {
return nil, virtual.ErrNotFound
}
lines, err := h.localGemsLines(ctx, repo, gemsIndexPrefix+name+"/")
if err != nil {
return nil, err
}
if len(lines[name]) == 0 {
return nil, virtual.ErrNotFound
}
return gems.InfoBody(lines[name]), nil
}
if path != "versions" && path != "names" {
return nil, virtual.ErrNotFound
}
lines, err := h.localGemsLines(ctx, repo, gemsIndexPrefix)
if err != nil {
return nil, err
}
names := slices.Sorted(maps.Keys(lines))
var b strings.Builder
if path == "names" {
b.WriteString("---\n")
for _, n := range names {
b.WriteString(n + "\n")
}
return []byte(b.String()), nil
}
b.WriteString(gems.VersionsHeader(time.Now()))
for _, n := range names {
b.WriteString(gems.VersionsLine(n, gems.InfoBody(lines[n])) + "\n")
}
return []byte(b.String()), nil
}
// localGemsLines returns /info lines per gem name in push order.
func (h *ProxyHandler) localGemsLines(ctx context.Context, repo, prefix string) (map[string][]string, error) {
files, err := h.db.ListLocalFilesByPrefix(ctx, repo, prefix)
if err != nil {
return nil, err
}
slices.SortFunc(files, func(a, b database.LocalFile) int { return cmp.Compare(a.ID, b.ID) })
out := map[string][]string{}
for _, f := range files {
// LIKE treats '_' as a wildcard, so the prefix match can over-select.
rest, ok := strings.CutPrefix(f.FilePath, prefix)
if !ok {
continue
}
name := strings.TrimPrefix(prefix, gemsIndexPrefix)
if name == "" {
name, _, _ = strings.Cut(rest, "/")
} else {
name = strings.TrimSuffix(name, "/")
}
line, err := h.readLocalBlob(ctx, f.ContentHash)
if err != nil {
return nil, err
}
out[name] = append(out[name], string(line))
}
return out, nil
}
+254
View File
@@ -0,0 +1,254 @@
package v1
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"
)
func testGemFile(t *testing.T, name, version, platform string) []byte {
t.Helper()
meta := fmt.Sprintf(`--- !ruby/object:Gem::Specification
name: %s
version: !ruby/object:Gem::Version
version: %s
platform: %s
dependencies:
- !ruby/object:Gem::Dependency
name: rack
requirement: !ruby/object:Gem::Requirement
requirements:
- - "~>"
- !ruby/object:Gem::Version
version: '3.0'
type: :runtime
prerelease: false
rubygems_version: 3.5.22
specification_version: 4
summary: test
`, name, version, platform)
var gz bytes.Buffer
zw := gzip.NewWriter(&gz)
_, _ = zw.Write([]byte(meta))
_ = zw.Close()
var buf bytes.Buffer
tw := tar.NewWriter(&buf)
_ = tw.WriteHeader(&tar.Header{Name: "metadata.gz", Mode: 0o444, Size: int64(gz.Len())})
_, _ = tw.Write(gz.Bytes())
_ = tw.Close()
return buf.Bytes()
}
func sha256Hex(b []byte) string {
s := sha256.Sum256(b)
return hex.EncodeToString(s[:])
}
func TestGemsPushIndexYank(t *testing.T) {
h, srv := npmTestServer(t)
repo := npmRepo(t, h, models.PackageGems)
base := srv.URL + "/api/v1/local/" + repo + "/"
g1 := testGemFile(t, "hello_gem", "0.1.0", "ruby")
g2 := testGemFile(t, "hello_gem", "0.2.0", "ruby")
gj := testGemFile(t, "hello_gem", "0.2.0", "java")
other := testGemFile(t, "hello-gem", "1.0.0", "ruby")
for _, g := range [][]byte{g1, g2, gj, other} {
if code, body := npmDo(t, http.MethodPost, base+"api/v1/gems", string(g)); code != http.StatusOK {
t.Fatalf("push = %d: %s", code, body)
}
}
if code, _ := npmDo(t, http.MethodPost, base+"api/v1/gems", string(g1)+"x"); code != http.StatusConflict {
t.Fatalf("repush with different bytes = %d, want 409", code)
}
if code, _ := npmDo(t, http.MethodPost, base+"api/v1/gems", string(g1)); code != http.StatusConflict {
t.Fatalf("identical repush = %d, want 409", code)
}
if code, _ := npmDo(t, http.MethodPost, base+"api/v1/gems", "not a gem"); code != http.StatusBadRequest {
t.Fatalf("garbage push = %d, want 400", code)
}
code, info := npmDo(t, http.MethodGet, base+"info/hello_gem", "")
want := "---\n0.1.0 rack:~> 3.0|checksum:" + sha256Hex(g1) + "\n0.2.0 rack:~> 3.0|checksum:" + sha256Hex(g2) +
"\n0.2.0-java rack:~> 3.0|checksum:" + sha256Hex(gj) + "\n"
if code != http.StatusOK || info != want {
t.Fatalf("info = %d\n%q\nwant\n%q", code, info, want)
}
if code, _ := npmDo(t, http.MethodHead, base+"versions", ""); code != http.StatusOK {
t.Fatalf("HEAD versions = %d; gem probes the compact index with HEAD", code)
}
_, versions := npmDo(t, http.MethodGet, base+"versions", "")
head, body, ok := strings.Cut(versions, "\n---\n")
if !ok || !strings.HasPrefix(head, "created_at: ") {
t.Fatalf("versions header = %q", versions)
}
_, otherInfo := npmDo(t, http.MethodGet, base+"info/hello-gem", "")
wantBody := fmt.Sprintf("hello-gem 1.0.0 %x\nhello_gem 0.1.0,0.2.0,0.2.0-java %x\n", md5.Sum([]byte(otherInfo)), md5.Sum([]byte(info)))
if body != wantBody {
t.Fatalf("versions body =\n%q\nwant\n%q", body, wantBody)
}
if _, names := npmDo(t, http.MethodGet, base+"names", ""); names != "---\nhello-gem\nhello_gem\n" {
t.Fatalf("names = %q", names)
}
if code, got := npmDo(t, http.MethodGet, base+"gems/hello_gem-0.2.0-java.gem", ""); code != http.StatusOK || got != string(gj) {
t.Fatalf("download = %d", code)
}
if code, _ := npmDo(t, http.MethodGet, base+"quick/Marshal.4.8/hello_gem-0.1.0.gemspec.rz", ""); code != http.StatusOK {
t.Fatalf("quick spec = %d", code)
}
if code, _ := npmDo(t, http.MethodGet, base+"info/missing", ""); code != http.StatusNotFound {
t.Fatalf("missing info = %d", code)
}
yank := func(form string) int {
req, _ := http.NewRequest(http.MethodDelete, base+"api/v1/gems/yank", strings.NewReader(form))
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatal(err)
}
_ = resp.Body.Close()
return resp.StatusCode
}
if c := yank("gem_name=hello_gem&version=0.2.0&platform=java"); c != http.StatusOK {
t.Fatalf("yank = %d", c)
}
if c := yank("gem_name=hello_gem&version=0.2.0&platform=java"); c != http.StatusNotFound {
t.Fatalf("second yank = %d, want 404", c)
}
if c := yank("gem_name=../x&version=1"); c != http.StatusBadRequest {
t.Fatalf("bad yank = %d, want 400", c)
}
if _, info := npmDo(t, http.MethodGet, base+"info/hello_gem", ""); strings.Contains(info, "java") || !strings.Contains(info, "0.2.0 ") {
t.Fatalf("info after yank = %q", info)
}
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 _, 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)
}
}
func TestGemsPushRejects(t *testing.T) {
h, srv := npmTestServer(t)
npmLocal := npmRepo(t, h, models.PackageNPM)
g := string(testGemFile(t, "g", "1.0.0", "ruby"))
if code, _ := npmDo(t, http.MethodPost, srv.URL+"/api/v1/local/"+npmLocal+"/api/v1/gems", g); code != http.StatusMethodNotAllowed {
t.Fatalf("push to npm local = %d, want 405", code)
}
repo := npmRepo(t, h, models.PackageGems)
if code, _ := npmDo(t, http.MethodPost, srv.URL+"/api/v1/local/"+repo+"/other", g); code != http.StatusNotFound {
t.Fatalf("push to other path = %d, want 404", code)
}
old := gemsMaxPushBytes
gemsMaxPushBytes = 64
defer func() { gemsMaxPushBytes = old }()
if code, _ := npmDo(t, http.MethodPost, srv.URL+"/api/v1/local/"+repo+"/api/v1/gems", g); code != http.StatusRequestEntityTooLarge {
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)
}
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
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", 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 {
t.Fatalf("push after release = %d: %s", code, body)
}
}
+39
View File
@@ -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))
}
}
+16 -2
View File
@@ -15,7 +15,9 @@ import (
"github.com/go-chi/chi/v5" "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/database"
"git.unkin.net/unkin/artifactapi/internal/proxy"
"git.unkin.net/unkin/artifactapi/internal/storage" "git.unkin.net/unkin/artifactapi/internal/storage"
"git.unkin.net/unkin/artifactapi/internal/testsupport" "git.unkin.net/unkin/artifactapi/internal/testsupport"
"git.unkin.net/unkin/artifactapi/internal/virtual" "git.unkin.net/unkin/artifactapi/internal/virtual"
@@ -25,6 +27,7 @@ import (
var ( var (
testDSN string testDSN string
testMinio testsupport.MinioConn testMinio testsupport.MinioConn
testRedis string
) )
func TestMain(m *testing.M) { func TestMain(m *testing.M) {
@@ -38,8 +41,15 @@ func TestMain(m *testing.M) {
termPG() termPG()
os.Exit(m.Run()) 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() code := m.Run()
termRedis()
termMinio() termMinio()
termPG() termPG()
os.Exit(code) os.Exit(code)
@@ -65,7 +75,11 @@ func npmTestServer(t *testing.T) (*ProxyHandler, *httptest.Server) {
if err != nil { if err != nil {
t.Fatal(err) 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 := chi.NewRouter()
r.Mount("/api/v1", h.Routes()) r.Mount("/api/v1", h.Routes())
srv := httptest.NewServer(r) srv := httptest.NewServer(r)
+19
View File
@@ -47,8 +47,12 @@ func (h *ProxyHandler) Routes() chi.Router {
r.Get("/remote/{remoteName}/*", h.handleProxy) r.Get("/remote/{remoteName}/*", h.handleProxy)
r.Head("/remote/{remoteName}/*", h.handleProxyHead) r.Head("/remote/{remoteName}/*", h.handleProxyHead)
r.Get("/local/{localName}/*", h.handleLocal) r.Get("/local/{localName}/*", h.handleLocal)
r.Head("/local/{localName}/*", h.handleLocal)
r.Put("/local/{localName}/*", h.handleLocalPut) r.Put("/local/{localName}/*", h.handleLocalPut)
r.Post("/local/{localName}/*", h.handleLocalPost)
r.Delete("/local/{localName}/*", h.handleLocalDelete)
r.Get("/virtual/{virtualName}/*", h.handleVirtual) r.Get("/virtual/{virtualName}/*", h.handleVirtual)
r.Head("/virtual/{virtualName}/*", h.handleVirtual)
return r return r
} }
@@ -201,6 +205,11 @@ func (h *ProxyHandler) handleVirtual(w http.ResponseWriter, r *http.Request) {
return 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) loc, ok, err := h.virtualEngine.MemberRedirect(r.Context(), *virt, path, r.URL.RawQuery, proxyBaseURL)
if err != nil { if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest) http.Error(w, err.Error(), http.StatusBadRequest)
@@ -254,6 +263,16 @@ func (h *ProxyHandler) handleLocal(w http.ResponseWriter, r *http.Request) {
return 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) prov, _ := provider.Get(remote.PackageType)
if indexer, ok := prov.(provider.LocalIndexer); ok { if indexer, ok := prov.(provider.LocalIndexer); ok {
+43 -7
View File
@@ -61,18 +61,54 @@ func TestTTL(t *testing.T) {
func TestLock(t *testing.T) { func TestLock(t *testing.T) {
requireRedis(t) requireRedis(t)
ctx := context.Background() ctx := context.Background()
ok, err := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute) token, err := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute)
if err != nil || !ok { if err != nil || token == "" {
t.Fatalf("first acquire should succeed: %v %v", ok, err) 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") 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) t.Fatal(err)
} }
if ok, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); !ok { if again, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); again != "" {
t.Error("acquire should succeed after release") 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")
} }
} }
+14 -5
View File
@@ -2,6 +2,7 @@ package cache
import ( import (
"context" "context"
"crypto/rand"
"fmt" "fmt"
"strings" "strings"
"time" "time"
@@ -46,15 +47,23 @@ func (r *Redis) CheckTTL(ctx context.Context, remote, path string) (bool, error)
return exists > 0, nil 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) key := fmt.Sprintf("lock:%s:%s", remote, path)
ok, err := r.client.SetNX(ctx, key, "1", ttl).Result() token := rand.Text()
return ok, err 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) 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. // SetETag stores etag for path, or drops the stored one when etag is empty.
+69
View File
@@ -207,3 +207,72 @@ func TestGitHubSyncLeaseFailureBackoff(t *testing.T) {
t.Fatal("claimed inside freshness window after a successful sync") t.Fatal("claimed inside freshness window after a successful sync")
} }
} }
// Deleting a remote drops its sync state, so a recreated remote of the same name
// starts clean instead of inheriting the old ETag, sync time and retry state.
func TestGitHubSyncStateDeletedWithRemote(t *testing.T) {
requireDB(t)
for _, tc := range []struct {
table string
pt models.PackageType
claim func(name string) (bool, string, error)
release func(name string, res provider.SyncResult) error
}{
{"github_rpm_sync_state", models.PackageGitHubRPM,
func(n string) (bool, string, error) {
return testDB.ClaimGitHubSyncLease(ctx(), n, "r1", time.Hour, time.Minute)
},
func(n string, res provider.SyncResult) error {
return testDB.ReleaseGitHubSyncLease(ctx(), n, "r1", res)
}},
{"github_deb_sync_state", models.PackageGitHubDeb,
func(n string) (bool, string, error) {
return testDB.ClaimGitHubDebSyncLease(ctx(), n, "r1", time.Hour, time.Minute)
},
func(n string, res provider.SyncResult) error {
return testDB.ReleaseGitHubDebSyncLease(ctx(), n, "r1", res)
}},
{"github_alpine_sync_state", models.PackageGitHubAlpine,
func(n string) (bool, string, error) {
return testDB.ClaimGitHubAlpineSyncLease(ctx(), n, "r1", time.Hour, time.Minute)
},
func(n string, res provider.SyncResult) error {
return testDB.ReleaseGitHubAlpineSyncLease(ctx(), n, "r1", res)
}},
} {
t.Run(tc.table, func(t *testing.T) {
name := "gh-recreate-" + time.Now().Format("150405.000000")
create := func() {
t.Helper()
if err := testDB.CreateRemote(ctx(), &models.Remote{
Name: name, PackageType: tc.pt, RepoType: models.RepoTypeRemote,
BaseURL: "https://api.github.com/repos/acme/tools", ReleasesRemote: "github", MutableTTL: 3600,
}); err != nil {
t.Fatalf("create remote: %v", err)
}
}
create()
if ok, _, err := tc.claim(name); err != nil || !ok {
t.Fatalf("claim: ok=%v err=%v", ok, err)
}
if err := tc.release(name, provider.SyncResult{Etag: `"old"`}); err != nil {
t.Fatalf("release: %v", err)
}
if err := testDB.DeleteRemote(ctx(), name); err != nil {
t.Fatalf("delete remote: %v", err)
}
var rows int
if err := testDB.Pool.QueryRow(ctx(), `SELECT count(*) FROM `+tc.table+` WHERE remote_name = $1`, name).Scan(&rows); err != nil || rows != 0 {
t.Fatalf("sync state left after delete: rows=%d err=%v", rows, err)
}
create()
ok, etag, err := tc.claim(name)
if err != nil || !ok || etag != "" {
t.Fatalf("recreated remote claim ok=%v etag=%q err=%v, want a fresh claim with no etag", ok, etag, err)
}
})
}
}
+14 -13
View File
@@ -372,6 +372,9 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
return newEtag, false, err return newEtag, false, err
} }
// A transiently failed asset fails the scan so the ETag is not advanced and
// the retry re-derives it; an invalid package is skipped as no retry fixes it.
var failed []error
seen := map[string]bool{} seen := map[string]bool{}
for _, rel := range releases { for _, rel := range releases {
if rel.Draft { if rel.Draft {
@@ -400,10 +403,14 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
meta, err := p.deriveAsset(ctx, remote, asset, fp) meta, err := p.deriveAsset(ctx, remote, asset, fp)
if err != nil { if err != nil {
slog.Warn("github_alpine: derive asset failed", "remote", remote.Name, "asset", asset.Name, "error", err) slog.Warn("github_alpine: derive asset failed", "remote", remote.Name, "asset", asset.Name, "error", err)
if !errors.Is(err, provider.ErrInvalidPackage) {
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
}
continue continue
} }
if err := inserter.InsertAlpineMetadata(ctx, meta); err != nil { if err := inserter.InsertAlpineMetadata(ctx, meta); err != nil {
slog.Error("github_alpine: insert metadata failed", "remote", remote.Name, "asset", asset.Name, "error", err) slog.Error("github_alpine: insert metadata failed", "remote", remote.Name, "asset", asset.Name, "error", err)
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
continue continue
} }
slog.Info("github_alpine: derived asset", "remote", remote.Name, "name", meta.Name, "version", meta.Version, "arch", meta.Arch) slog.Info("github_alpine: derived asset", "remote", remote.Name, "name", meta.Name, "version", meta.Version, "arch", meta.Arch)
@@ -415,7 +422,7 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
_ = deleter.DeleteAlpineMetadata(ctx, remote.Name, fp) _ = deleter.DeleteAlpineMetadata(ctx, remote.Name, fp)
} }
} }
return newEtag, true, nil return newEtag, true, errors.Join(failed...)
} }
type ghRelease struct { type ghRelease struct {
@@ -496,7 +503,7 @@ func (p *GitHubProvider) deriveAsset(ctx context.Context, remote models.Remote,
return nil, err return nil, err
} }
if meta.Name == "" || meta.Arch == "" { if meta.Name == "" || meta.Arch == "" {
return nil, errors.New(".PKGINFO missing pkgname/arch") return nil, fmt.Errorf("%w: .PKGINFO missing pkgname/arch", provider.ErrInvalidPackage)
} }
meta.RepoName = remote.Name meta.RepoName = remote.Name
@@ -526,13 +533,13 @@ func (p *GitHubProvider) fetchPkginfo(ctx context.Context, remote models.Remote,
} }
meta, complete, perr := pkginfoFromPrefix(body) meta, complete, perr := pkginfoFromPrefix(body)
if perr != nil { if perr != nil {
return nil, fmt.Errorf("parse apk .PKGINFO: %w", perr) return nil, fmt.Errorf("%w: parse apk .PKGINFO: %w", provider.ErrInvalidPackage, perr)
} }
if complete { if complete {
return meta, nil return meta, nil
} }
if full || n >= p.headerMax { if full || n >= p.headerMax {
return nil, fmt.Errorf(".PKGINFO not found within %d bytes of %s", n, downloadURL) return nil, fmt.Errorf("%w: .PKGINFO not found within %d bytes of %s", provider.ErrInvalidPackage, n, downloadURL)
} }
n *= 2 n *= 2
if n > p.headerMax { if n > p.headerMax {
@@ -594,7 +601,7 @@ func pkginfoFromPrefix(prefix []byte) (meta *provider.AlpineMetadata, complete b
} }
// rangeGet returns the first n bytes of downloadURL. full is true when the // rangeGet returns the first n bytes of downloadURL. full is true when the
// response body was shorter than n (i.e. we already have the whole object). // response holds the whole object.
func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, downloadURL string, n int64) ([]byte, bool, error) { func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, downloadURL string, n int64) ([]byte, bool, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil) req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
if err != nil { if err != nil {
@@ -616,15 +623,9 @@ func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, dow
} }
defer resp.Body.Close() defer resp.Body.Close()
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent { if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
return nil, false, fmt.Errorf("range GET %s: status %d", downloadURL, resp.StatusCode) return nil, false, fmt.Errorf("range GET: %w", provider.AssetStatusError(downloadURL, resp))
} }
return provider.ReadPrefix(resp, n)
body, err := io.ReadAll(io.LimitReader(resp.Body, n))
if err != nil {
return nil, false, err
}
full := int64(len(body)) < n
return body, full, nil
} }
// assetPath is the package's location relative to github.com — the path the // assetPath is the package's location relative to github.com — the path the
+22 -3
View File
@@ -8,6 +8,7 @@ import (
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"io" "io"
"net/http" "net/http"
@@ -88,15 +89,17 @@ type githubFixture struct {
notModHit int notModHit int
releaseAuth string releaseAuth string
assetAuth string assetAuth string
assetFail map[string]int // asset filename -> downloads left to answer 500
mu sync.Mutex mu sync.Mutex
} }
func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture { func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
t.Helper() t.Helper()
f := &githubFixture{ f := &githubFixture{
apkBytes: map[string][]byte{}, apkBytes: map[string][]byte{},
rangeHit: map[string]int{}, rangeHit: map[string]int{},
fullHit: map[string]int{}, fullHit: map[string]int{},
assetFail: map[string]int{},
} }
f.apkBytes["demo-1.2.3-r0.apk"] = testsupport.MinimalApk("demo", "1.2.3-r0", "x86_64") f.apkBytes["demo-1.2.3-r0.apk"] = testsupport.MinimalApk("demo", "1.2.3-r0", "x86_64")
@@ -146,6 +149,12 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
} }
rng := r.Header.Get("Range") rng := r.Header.Get("Range")
f.mu.Lock() f.mu.Lock()
if f.assetFail[name] > 0 {
f.assetFail[name]--
f.mu.Unlock()
http.Error(w, "boom", http.StatusInternalServerError)
return
}
f.assetAuth = r.Header.Get("Authorization") f.assetAuth = r.Header.Get("Authorization")
if rng != "" { if rng != "" {
f.rangeHit[name]++ f.rangeHit[name]++
@@ -495,3 +504,13 @@ func readAPKIndex(t *testing.T, gzBytes []byte) string {
} }
} }
} }
func TestGitHubDeriveAssetMissingPkgnameIsInvalid(t *testing.T) {
fx := newGitHubFixture(t, true)
fx.apkBytes["nameless-1-r0.apk"] = testsupport.MinimalApk("", "1-r0", "x86_64")
asset := ghAsset{BrowserDownloadURL: fx.srv.URL + "/acme/tools/releases/download/v1.2.3/nameless-1-r0.apk"}
_, err := newTestProvider().deriveAsset(context.Background(), fx.remote(), asset, "fp")
if !errors.Is(err, provider.ErrInvalidPackage) {
t.Fatalf("want ErrInvalidPackage, got %v", err)
}
}
+55
View File
@@ -308,3 +308,58 @@ func TestSyncerPrimeBypassesRecencyPeriodicDoesNot(t *testing.T) {
t.Fatalf("periodic scan ran inside recency window: %d -> %d releases calls", releasesAfterPrime, fx.releasesHit) t.Fatalf("periodic scan ran inside recency window: %d -> %d releases calls", releasesAfterPrime, fx.releasesHit)
} }
} }
// A transiently failed asset fails the scan: the ETag is not advanced, a retry
// is scheduled, and the retry against the unchanged release list derives it. An
// invalid package is skipped instead, since no retry can fix it.
func TestSyncerFailedAssetRetry(t *testing.T) {
for _, tc := range []struct {
name string
corrupt bool
wantRetry bool
}{
{name: "transient", wantRetry: true},
{name: "invalid package", corrupt: true},
} {
t.Run(tc.name, func(t *testing.T) {
fx := newGitHubFixture(t, true)
fx.etag = `"v1"`
if tc.corrupt {
fx.apkBytes["other-9-r0.apk"] = []byte("not a package")
} else {
fx.apkBytes["other-9-r0.apk"] = testsupport.MinimalApk("other", "9-r0", "aarch64")
fx.assetFail["other-9-r0.apk"] = 1
}
store := newFakeSyncStore()
s := newSyncer(store, newTestProvider(), testSyncConfig())
remote := fx.remote()
bg := context.Background()
s.process(bg, syncJob{remote: remote, prime: true})
_, pending := store.retryAt[remote.Name]
if pending != tc.wantRetry {
t.Fatalf("retry pending=%v, want %v", pending, tc.wantRetry)
}
if _, saved := store.etags[remote.Name]; saved == tc.wantRetry {
t.Fatalf("etag saved=%v after first scan", saved)
}
if rows, _ := store.ListAlpineMetadataEntries(bg, remote.Name); len(rows) != 1 {
t.Fatalf("healthy asset not served: %d rows", len(rows))
}
if !tc.wantRetry {
return
}
store.mu.Lock()
store.retryAt[remote.Name] = time.Now().Add(-time.Second)
store.mu.Unlock()
s.process(bg, syncJob{remote: remote})
if _, pending := store.retryAt[remote.Name]; pending || store.etags[remote.Name] != `"v1"` {
t.Fatalf("retry not recorded: pending=%v etag=%q", pending, store.etags[remote.Name])
}
if rows, _ := store.ListAlpineMetadataEntries(bg, remote.Name); len(rows) != 2 {
t.Fatalf("retry did not derive the failed asset: %d rows", len(rows))
}
})
}
}
+15 -14
View File
@@ -333,6 +333,9 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
return newEtag, false, err return newEtag, false, err
} }
// A transiently failed asset fails the scan so the ETag is not advanced and
// the retry re-derives it; an invalid package is skipped as no retry fixes it.
var failed []error
seen := map[string]bool{} seen := map[string]bool{}
for _, rel := range releases { for _, rel := range releases {
if rel.Draft { if rel.Draft {
@@ -361,10 +364,14 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
meta, err := p.deriveAsset(ctx, remote, asset, fp) meta, err := p.deriveAsset(ctx, remote, asset, fp)
if err != nil { if err != nil {
slog.Warn("github_deb: derive asset failed", "remote", remote.Name, "asset", asset.Name, "error", err) slog.Warn("github_deb: derive asset failed", "remote", remote.Name, "asset", asset.Name, "error", err)
if !errors.Is(err, provider.ErrInvalidPackage) {
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
}
continue continue
} }
if err := store.InsertDebMetadata(ctx, meta); err != nil { if err := store.InsertDebMetadata(ctx, meta); err != nil {
slog.Error("github_deb: insert metadata failed", "remote", remote.Name, "asset", asset.Name, "error", err) slog.Error("github_deb: insert metadata failed", "remote", remote.Name, "asset", asset.Name, "error", err)
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
continue continue
} }
slog.Info("github_deb: derived asset", "remote", remote.Name, "name", meta.Name, "version", meta.Version, "arch", meta.Architecture) slog.Info("github_deb: derived asset", "remote", remote.Name, "name", meta.Name, "version", meta.Version, "arch", meta.Architecture)
@@ -376,7 +383,7 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
_ = store.DeleteDebMetadata(ctx, remote.Name, fp) _ = store.DeleteDebMetadata(ctx, remote.Name, fp)
} }
} }
return newEtag, true, nil return newEtag, true, errors.Join(failed...)
} }
type ghRelease struct { type ghRelease struct {
@@ -468,7 +475,7 @@ func (p *GitHubProvider) deriveAsset(ctx context.Context, remote models.Remote,
Size: asset.Size, Size: asset.Size,
} }
if meta.Name == "" { if meta.Name == "" {
return nil, errors.New("control missing Package field") return nil, fmt.Errorf("%w: control missing Package field", provider.ErrInvalidPackage)
} }
// The Packages SHA256 must be the sha256 of the whole .deb. Prefer GitHub's // The Packages SHA256 must be the sha256 of the whole .deb. Prefer GitHub's
@@ -503,13 +510,13 @@ func (p *GitHubProvider) fetchControl(ctx context.Context, remote models.Remote,
} }
control, complete, perr := controlFromPrefix(body) control, complete, perr := controlFromPrefix(body)
if perr != nil { if perr != nil {
return "", fmt.Errorf("parse deb control: %w", perr) return "", fmt.Errorf("%w: parse deb control: %w", provider.ErrInvalidPackage, perr)
} }
if complete { if complete {
return control, nil return control, nil
} }
if full || n >= p.headerMax { if full || n >= p.headerMax {
return "", fmt.Errorf("control.tar not found within %d bytes of %s", n, downloadURL) return "", fmt.Errorf("%w: control.tar not found within %d bytes of %s", provider.ErrInvalidPackage, n, downloadURL)
} }
n *= 2 n *= 2
if n > p.headerMax { if n > p.headerMax {
@@ -567,7 +574,7 @@ func controlFromPrefix(prefix []byte) (control string, complete bool, err error)
} }
// rangeGet returns the first n bytes of downloadURL. full is true when the // rangeGet returns the first n bytes of downloadURL. full is true when the
// response body was shorter than n (i.e. we already have the whole object). // response holds the whole object.
func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, downloadURL string, n int64) ([]byte, bool, error) { func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, downloadURL string, n int64) ([]byte, bool, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil) req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
if err != nil { if err != nil {
@@ -589,15 +596,9 @@ func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, dow
} }
defer resp.Body.Close() defer resp.Body.Close()
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent { if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
return nil, false, fmt.Errorf("range GET %s: status %d", downloadURL, resp.StatusCode) return nil, false, fmt.Errorf("range GET: %w", provider.AssetStatusError(downloadURL, resp))
} }
return provider.ReadPrefix(resp, n)
body, err := io.ReadAll(io.LimitReader(resp.Body, n))
if err != nil {
return nil, false, err
}
full := int64(len(body)) < n
return body, full, nil
} }
func (p *GitHubProvider) computeSHA256(ctx context.Context, remote models.Remote, downloadURL string) (string, error) { func (p *GitHubProvider) computeSHA256(ctx context.Context, remote models.Remote, downloadURL string) (string, error) {
@@ -620,7 +621,7 @@ func (p *GitHubProvider) computeSHA256(ctx context.Context, remote models.Remote
} }
defer resp.Body.Close() defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { if resp.StatusCode != http.StatusOK {
return "", fmt.Errorf("GET %s: status %d", downloadURL, resp.StatusCode) return "", fmt.Errorf("GET: %w", provider.AssetStatusError(downloadURL, resp))
} }
h := sha256.New() h := sha256.New()
+22 -3
View File
@@ -7,6 +7,7 @@ import (
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"io" "io"
"net/http" "net/http"
@@ -81,15 +82,17 @@ type githubFixture struct {
notModHit int notModHit int
releaseAuth string releaseAuth string
assetAuth string assetAuth string
assetFail map[string]int // asset filename -> downloads left to answer 500
mu sync.Mutex mu sync.Mutex
} }
func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture { func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
t.Helper() t.Helper()
f := &githubFixture{ f := &githubFixture{
debBytes: map[string][]byte{}, debBytes: map[string][]byte{},
rangeHit: map[string]int{}, rangeHit: map[string]int{},
fullHit: map[string]int{}, fullHit: map[string]int{},
assetFail: map[string]int{},
} }
f.debBytes["demo_1.2-3_amd64.deb"] = testsupport.MinimalDeb("demo", "1.2-3", "amd64") f.debBytes["demo_1.2-3_amd64.deb"] = testsupport.MinimalDeb("demo", "1.2-3", "amd64")
@@ -139,6 +142,12 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
} }
rng := r.Header.Get("Range") rng := r.Header.Get("Range")
f.mu.Lock() f.mu.Lock()
if f.assetFail[name] > 0 {
f.assetFail[name]--
f.mu.Unlock()
http.Error(w, "boom", http.StatusInternalServerError)
return
}
f.assetAuth = r.Header.Get("Authorization") f.assetAuth = r.Header.Get("Authorization")
if rng != "" { if rng != "" {
f.rangeHit[name]++ f.rangeHit[name]++
@@ -460,3 +469,13 @@ func TestGitHubAssetPatternFilter(t *testing.T) {
t.Fatalf("pattern filter failed, rows=%+v", rows) t.Fatalf("pattern filter failed, rows=%+v", rows)
} }
} }
func TestGitHubDeriveAssetMissingPackageIsInvalid(t *testing.T) {
fx := newGitHubFixture(t, true)
fx.debBytes["nameless_1_amd64.deb"] = testsupport.MinimalDeb("", "1", "amd64")
asset := ghAsset{BrowserDownloadURL: fx.srv.URL + "/acme/tools/releases/download/v1.2-3/nameless_1_amd64.deb", Digest: "sha256:00"}
_, err := newTestProvider().deriveAsset(context.Background(), fx.remote(), asset, "fp")
if !errors.Is(err, provider.ErrInvalidPackage) {
t.Fatalf("want ErrInvalidPackage, got %v", err)
}
}
+55
View File
@@ -308,3 +308,58 @@ func TestSyncerPrimeBypassesRecencyPeriodicDoesNot(t *testing.T) {
t.Fatalf("periodic scan ran inside recency window: %d -> %d releases calls", releasesAfterPrime, fx.releasesHit) t.Fatalf("periodic scan ran inside recency window: %d -> %d releases calls", releasesAfterPrime, fx.releasesHit)
} }
} }
// A transiently failed asset fails the scan: the ETag is not advanced, a retry
// is scheduled, and the retry against the unchanged release list derives it. An
// invalid package is skipped instead, since no retry can fix it.
func TestSyncerFailedAssetRetry(t *testing.T) {
for _, tc := range []struct {
name string
corrupt bool
wantRetry bool
}{
{name: "transient", wantRetry: true},
{name: "invalid package", corrupt: true},
} {
t.Run(tc.name, func(t *testing.T) {
fx := newGitHubFixture(t, true)
fx.etag = `"v1"`
if tc.corrupt {
fx.debBytes["other_9-9_arm64.deb"] = []byte("not a package")
} else {
fx.debBytes["other_9-9_arm64.deb"] = testsupport.MinimalDeb("other", "9-9", "arm64")
fx.assetFail["other_9-9_arm64.deb"] = 1
}
store := newFakeSyncStore()
s := newSyncer(store, newTestProvider(), testSyncConfig())
remote := fx.remote()
bg := context.Background()
s.process(bg, syncJob{remote: remote, prime: true})
_, pending := store.retryAt[remote.Name]
if pending != tc.wantRetry {
t.Fatalf("retry pending=%v, want %v", pending, tc.wantRetry)
}
if _, saved := store.etags[remote.Name]; saved == tc.wantRetry {
t.Fatalf("etag saved=%v after first scan", saved)
}
if rows, _ := store.ListDebMetadataEntries(bg, remote.Name); len(rows) != 1 {
t.Fatalf("healthy asset not served: %d rows", len(rows))
}
if !tc.wantRetry {
return
}
store.mu.Lock()
store.retryAt[remote.Name] = time.Now().Add(-time.Second)
store.mu.Unlock()
s.process(bg, syncJob{remote: remote})
if _, pending := store.retryAt[remote.Name]; pending || store.etags[remote.Name] != `"v1"` {
t.Fatalf("retry not recorded: pending=%v etag=%q", pending, store.etags[remote.Name])
}
if rows, _ := store.ListDebMetadataEntries(bg, remote.Name); len(rows) != 2 {
t.Fatalf("retry did not derive the failed asset: %d rows", len(rows))
}
})
}
}
+468
View File
@@ -0,0 +1,468 @@
package gems
import (
"archive/tar"
"bytes"
"compress/gzip"
"compress/zlib"
"crypto/md5"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"io"
"maps"
"regexp"
"slices"
"strings"
"time"
"gopkg.in/yaml.v3"
)
var (
NameRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]*$`)
VersionRe = regexp.MustCompile(`^[0-9]+(\.[0-9A-Za-z]+)*(-[0-9A-Za-z-]+(\.[0-9A-Za-z-]+)*)?$`)
PlatformRe = regexp.MustCompile(`^[A-Za-z0-9_.-]+$`)
opRe = regexp.MustCompile(`^(=|!=|>|<|>=|<=|~>)$`)
)
const maxMetadataBytes = 8 << 20
type Requirement [][2]string
type Dependency struct {
Name string
Requirement Requirement
Type string
Prerelease bool
}
// Spec is the subset of a gemspec the compact index and quick spec need.
type Spec struct {
Name, Version, Platform string
Summary, RubyGemsVer string
SpecVersion int
Date time.Time
Authors, Licenses []string
Email, Description string
Homepage string
Metadata map[string]string
Dependencies []Dependency
RequiredRuby Requirement
RequiredRubyGems Requirement
}
// FullName is the gem's file stem: name-version, plus -platform when not ruby.
func (s *Spec) FullName() string {
if s.Platform == "" || s.Platform == "ruby" {
return s.Name + "-" + s.Version
}
return s.Name + "-" + s.Version + "-" + s.Platform
}
type yamlReq struct {
Requirements [][]yaml.Node `yaml:"requirements"`
}
func (r *yamlReq) decode() (Requirement, error) {
if r == nil {
return nil, nil
}
var out Requirement
for _, pair := range r.Requirements {
var op string
var ver struct {
Version string `yaml:"version"`
}
if len(pair) != 2 || pair[0].Decode(&op) != nil || pair[1].Decode(&ver) != nil {
return nil, errors.New("malformed requirement")
}
if !opRe.MatchString(op) || !VersionRe.MatchString(ver.Version) {
return nil, fmt.Errorf("invalid requirement %q %q", op, ver.Version)
}
out = append(out, [2]string{op, ver.Version})
}
return out, nil
}
// String renders the requirement as the compact index does: "op ver&op ver".
func (r Requirement) String() string {
parts := make([]string, len(r))
for i, p := range r {
parts[i] = p[0] + " " + p[1]
}
return strings.Join(parts, "&")
}
func (r Requirement) trivial() bool {
return len(r) == 0 || len(r) == 1 && r[0] == [2]string{">=", "0"}
}
// ParseGem reads the gemspec from a .gem archive (a tar holding metadata.gz).
func ParseGem(gem []byte) (*Spec, error) {
tr := tar.NewReader(bytes.NewReader(gem))
for {
hdr, err := tr.Next()
if err == io.EOF {
return nil, errors.New("gem has no metadata.gz")
}
if err != nil {
return nil, fmt.Errorf("read gem: %w", err)
}
if hdr.Name != "metadata.gz" {
continue
}
zr, err := gzip.NewReader(tr)
if err != nil {
return nil, fmt.Errorf("metadata.gz: %w", err)
}
raw, err := io.ReadAll(io.LimitReader(zr, maxMetadataBytes+1))
if err != nil {
return nil, fmt.Errorf("metadata.gz: %w", err)
}
if len(raw) > maxMetadataBytes {
return nil, fmt.Errorf("metadata exceeds %d bytes", maxMetadataBytes)
}
return parseMetadata(raw)
}
}
func parseMetadata(raw []byte) (*Spec, error) {
var doc struct {
Name string `yaml:"name"`
Version struct {
Version string `yaml:"version"`
} `yaml:"version"`
Platform string `yaml:"platform"`
Summary string `yaml:"summary"`
Authors []string `yaml:"authors"`
Email yaml.Node `yaml:"email"`
Description string `yaml:"description"`
Homepage string `yaml:"homepage"`
Licenses []string `yaml:"licenses"`
Metadata map[string]string `yaml:"metadata"`
Date string `yaml:"date"`
RubyGemsVer string `yaml:"rubygems_version"`
SpecVersion int `yaml:"specification_version"`
Dependencies []struct {
Name string `yaml:"name"`
Requirement *yamlReq `yaml:"requirement"`
VersionReqs *yamlReq `yaml:"version_requirements"`
Type string `yaml:"type"`
Prerelease bool `yaml:"prerelease"`
} `yaml:"dependencies"`
RequiredRuby *yamlReq `yaml:"required_ruby_version"`
RequiredRubyGems *yamlReq `yaml:"required_rubygems_version"`
}
if err := yaml.Unmarshal(raw, &doc); err != nil {
return nil, fmt.Errorf("gemspec yaml: %w", err)
}
s := &Spec{
Name: doc.Name, Version: doc.Version.Version, Platform: doc.Platform,
Summary: doc.Summary, Authors: doc.Authors, Description: doc.Description,
Homepage: doc.Homepage, Licenses: doc.Licenses, Metadata: doc.Metadata,
RubyGemsVer: doc.RubyGemsVer, SpecVersion: doc.SpecVersion,
}
if s.Platform == "" {
s.Platform = "ruby"
}
if !NameRe.MatchString(s.Name) || !VersionRe.MatchString(s.Version) || !PlatformRe.MatchString(s.Platform) {
return nil, fmt.Errorf("invalid gem name/version/platform %q %q %q", s.Name, s.Version, s.Platform)
}
if doc.Email.Kind == yaml.ScalarNode {
s.Email = doc.Email.Value
}
if t, err := time.Parse("2006-01-02", strings.SplitN(doc.Date, " ", 2)[0]); err == nil {
s.Date = t
}
var err error
if s.RequiredRuby, err = doc.RequiredRuby.decode(); err != nil {
return nil, err
}
if s.RequiredRubyGems, err = doc.RequiredRubyGems.decode(); err != nil {
return nil, err
}
for _, d := range doc.Dependencies {
if !NameRe.MatchString(d.Name) {
return nil, fmt.Errorf("invalid dependency name %q", d.Name)
}
req := d.Requirement
if req == nil {
req = d.VersionReqs
}
r, err := req.decode()
if err != nil {
return nil, fmt.Errorf("dependency %s: %w", d.Name, err)
}
typ := strings.TrimPrefix(d.Type, ":")
if typ == "" {
typ = "runtime"
}
s.Dependencies = append(s.Dependencies, Dependency{Name: d.Name, Requirement: r, Type: typ, Prerelease: d.Prerelease})
}
return s, nil
}
// InfoLine renders the spec's compact index /info line; checksum is the
// .gem's sha256 hex.
func InfoLine(s *Spec, gem []byte) string {
sum := sha256.Sum256(gem)
var deps []string
for _, d := range s.Dependencies {
if d.Type == "runtime" {
r := d.Requirement
if len(r) == 0 {
r = Requirement{{">=", "0"}}
}
deps = append(deps, d.Name+":"+r.String())
}
}
key := s.Version
if s.Platform != "ruby" {
key += "-" + s.Platform
}
line := key + " " + strings.Join(deps, ",") + "|checksum:" + hex.EncodeToString(sum[:])
if !s.RequiredRuby.trivial() {
line += ",ruby:" + s.RequiredRuby.String()
}
if !s.RequiredRubyGems.trivial() {
line += ",rubygems:" + s.RequiredRubyGems.String()
}
return line
}
// InfoBody renders an /info document from its lines.
func InfoBody(lines []string) []byte {
var b strings.Builder
b.WriteString("---\n")
for _, l := range lines {
b.WriteString(l)
b.WriteByte('\n')
}
return []byte(b.String())
}
// InfoKey returns an /info line's version[-platform] token.
func InfoKey(line string) string {
k, _, _ := strings.Cut(line, " ")
return k
}
// InfoLines returns an /info body's version lines.
func InfoLines(body []byte) []string {
var out []string
for _, l := range strings.Split(string(body), "\n") {
if l != "" && l != "---" {
out = append(out, l)
}
}
return out
}
func MD5Hex(b []byte) string {
sum := md5.Sum(b)
return hex.EncodeToString(sum[:])
}
// VersionsLine renders one /versions entry from a gem's /info body.
func VersionsLine(name string, info []byte) string {
var keys []string
for _, l := range InfoLines(info) {
keys = append(keys, InfoKey(l))
}
return name + " " + strings.Join(keys, ",") + " " + MD5Hex(info)
}
// VersionsHeader starts a /versions document.
func VersionsHeader(at time.Time) string {
return "created_at: " + at.UTC().Format(time.RFC3339) + "\n---\n"
}
// QuickSpec returns the deflated Marshal.4.8 dump of the spec, as served at
// quick/Marshal.4.8/<full>.gemspec.rz.
func QuickSpec(s *Spec) ([]byte, error) {
var buf bytes.Buffer
zw := zlib.NewWriter(&buf)
if _, err := zw.Write(MarshalSpec(s)); err != nil {
return nil, err
}
if err := zw.Close(); err != nil {
return nil, err
}
return buf.Bytes(), nil
}
// MarshalSpec mirrors Gem::Specification#_dump wrapped by Marshal.dump.
func MarshalSpec(s *Spec) []byte {
var in rbMarshal
in.WriteString("\x04\x08")
in.array(19)
in.str(s.RubyGemsVer)
in.WriteByte('i')
in.fixnum(s.SpecVersion)
in.str(s.Name)
in.version(s.Version)
in.time(s.Date)
in.str(s.Summary)
in.requirement(s.RequiredRuby)
in.requirement(s.RequiredRubyGems)
in.str(s.Platform)
in.array(len(s.Dependencies))
for _, d := range s.Dependencies {
in.WriteByte('o')
in.sym("Gem::Dependency")
in.fixnum(4)
in.sym("@name")
in.str(d.Name)
in.sym("@requirement")
in.requirement(d.Requirement)
in.sym("@type")
in.sym(d.Type)
in.sym("@prerelease")
in.boolean(d.Prerelease)
}
in.str("")
in.optStr(s.Email)
in.strs(s.Authors)
in.optStr(s.Description)
in.optStr(s.Homepage)
in.boolean(true)
in.str(s.Platform)
in.strs(s.Licenses)
in.WriteByte('{')
in.fixnum(len(s.Metadata))
for _, k := range slices.Sorted(maps.Keys(s.Metadata)) {
in.str(k)
in.str(s.Metadata[k])
}
var out rbMarshal
out.WriteString("\x04\x08u")
out.sym("Gem::Specification")
out.fixnum(in.Len())
out.Write(in.Bytes())
return out.Bytes()
}
// rbMarshal writes the Ruby Marshal 4.8 subset gemspecs use. It never emits
// object or symbol links, so every symbol is written in full.
type rbMarshal struct{ bytes.Buffer }
func (m *rbMarshal) fixnum(n int) {
switch {
case n == 0:
m.WriteByte(0)
case n > 0 && n < 123:
m.WriteByte(byte(n + 5))
case n < 0 && n > -124:
m.WriteByte(byte(n - 5))
default:
var b []byte
for x := n; len(b) < 4; x >>= 8 {
b = append(b, byte(x))
if (n >= 0 && x>>8 == 0) || (n < 0 && x>>8 == -1) {
break
}
}
if n >= 0 {
m.WriteByte(byte(len(b)))
} else {
m.WriteByte(byte(-len(b)))
}
m.Write(b)
}
}
func (m *rbMarshal) raw(s string) {
m.fixnum(len(s))
m.WriteString(s)
}
func (m *rbMarshal) sym(s string) {
m.WriteByte(':')
m.raw(s)
}
// str writes a UTF-8 String (an ivar-wrapped string with E=true).
func (m *rbMarshal) str(s string) {
m.WriteString("I\"")
m.raw(s)
m.fixnum(1)
m.sym("E")
m.WriteByte('T')
}
func (m *rbMarshal) optStr(s string) {
if s == "" {
m.WriteByte('0')
return
}
m.str(s)
}
func (m *rbMarshal) strs(ss []string) {
m.array(len(ss))
for _, s := range ss {
m.str(s)
}
}
func (m *rbMarshal) array(n int) {
m.WriteByte('[')
m.fixnum(n)
}
func (m *rbMarshal) boolean(b bool) {
if b {
m.WriteByte('T')
} else {
m.WriteByte('F')
}
}
func (m *rbMarshal) version(v string) {
m.WriteByte('U')
m.sym("Gem::Version")
m.array(1)
m.str(v)
}
func (m *rbMarshal) requirement(r Requirement) {
if len(r) == 0 {
r = Requirement{{">=", "0"}}
}
m.WriteByte('U')
m.sym("Gem::Requirement")
m.array(1)
m.array(len(r))
for _, p := range r {
m.array(2)
m.str(p[0])
m.version(p[1])
}
}
// time writes a UTC Time with day resolution, as Time#_dump lays it out.
func (m *rbMarshal) time(t time.Time) {
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")
m.sym("Time")
m.fixnum(8)
for _, w := range []uint32{p, s} {
m.Write([]byte{byte(w), byte(w >> 8), byte(w >> 16), byte(w >> 24)})
}
m.fixnum(1)
m.sym("zone")
m.WriteString("I\"")
m.raw("UTC")
m.fixnum(1)
m.sym("E")
m.WriteByte('F')
}
+223
View File
@@ -0,0 +1,223 @@
package gems
import (
"archive/tar"
"bytes"
"compress/gzip"
"compress/zlib"
"crypto/sha256"
"encoding/hex"
"io"
"strings"
"testing"
"time"
)
const testMetadata = `--- !ruby/object:Gem::Specification
name: unkin-e2e-hello
version: !ruby/object:Gem::Version
version: 0.1.0
platform: %PLATFORM%
authors:
- e2e
date: 2026-10-17 00:00:00.000000000 Z
dependencies:
- !ruby/object:Gem::Dependency
name: rack
requirement: &1 !ruby/object:Gem::Requirement
requirements:
- - ">="
- !ruby/object:Gem::Version
version: '2.0'
- - "<"
- !ruby/object:Gem::Version
version: '4'
type: :runtime
prerelease: false
version_requirements: *1
- !ruby/object:Gem::Dependency
name: json
version_requirements: !ruby/object:Gem::Requirement
requirements:
- - "~>"
- !ruby/object:Gem::Version
version: '2.1'
type: :runtime
prerelease: false
- !ruby/object:Gem::Dependency
name: rake
requirement: !ruby/object:Gem::Requirement
requirements:
- - ">="
- !ruby/object:Gem::Version
version: '0'
type: :development
prerelease: false
email:
licenses:
- MIT
metadata:
source_code_uri: https://example.com
required_ruby_version: !ruby/object:Gem::Requirement
requirements:
- - ">="
- !ruby/object:Gem::Version
version: '2.7'
required_rubygems_version: !ruby/object:Gem::Requirement
requirements:
- - ">="
- !ruby/object:Gem::Version
version: '0'
rubygems_version: 3.5.22
specification_version: 4
summary: hi
`
func buildGem(t *testing.T, metadata []byte) []byte {
t.Helper()
var gz bytes.Buffer
zw := gzip.NewWriter(&gz)
_, _ = zw.Write(metadata)
_ = zw.Close()
var buf bytes.Buffer
tw := tar.NewWriter(&buf)
for name, body := range map[string][]byte{"data.tar.gz": []byte("x"), "metadata.gz": gz.Bytes()} {
if err := tw.WriteHeader(&tar.Header{Name: name, Mode: 0o444, Size: int64(len(body))}); err != nil {
t.Fatal(err)
}
_, _ = tw.Write(body)
}
_ = tw.Close()
return buf.Bytes()
}
func testGem(t *testing.T, platform string) []byte {
return buildGem(t, []byte(strings.Replace(testMetadata, "%PLATFORM%", platform, 1)))
}
func TestParseGemAndInfoLine(t *testing.T) {
gem := testGem(t, "ruby")
s, err := ParseGem(gem)
if err != nil {
t.Fatal(err)
}
if s.FullName() != "unkin-e2e-hello-0.1.0" || !s.Date.Equal(time.Date(2026, 10, 17, 0, 0, 0, 0, time.UTC)) {
t.Fatalf("spec = %+v", s)
}
if len(s.Dependencies) != 3 || s.Dependencies[1].Requirement.String() != "~> 2.1" {
t.Fatalf("deps = %+v", s.Dependencies)
}
sum := sha256.Sum256(gem)
want := "0.1.0 rack:>= 2.0&< 4,json:~> 2.1|checksum:" + hex.EncodeToString(sum[:]) + ",ruby:>= 2.7"
if got := InfoLine(s, gem); got != want {
t.Fatalf("InfoLine =\n %s\nwant\n %s", got, want)
}
plat, err := ParseGem(testGem(t, "x86_64-linux"))
if err != nil {
t.Fatal(err)
}
if plat.FullName() != "unkin-e2e-hello-0.1.0-x86_64-linux" || InfoKey(InfoLine(plat, nil)) != "0.1.0-x86_64-linux" {
t.Fatalf("platform gem: %s / %s", plat.FullName(), InfoLine(plat, nil))
}
}
func TestParseGemRejects(t *testing.T) {
big := []byte(testMetadata + "description: " + strings.Repeat("a", maxMetadataBytes) + "\n")
cases := map[string][]byte{
"not a tar": []byte("garbage"),
"no metadata": func() []byte { var b bytes.Buffer; tw := tar.NewWriter(&b); _ = tw.Close(); return b.Bytes() }(),
"bad name": buildGem(t, []byte(strings.Replace(testMetadata, "name: unkin-e2e-hello", "name: ../x", 1))),
"bad version": buildGem(t, []byte(strings.Replace(testMetadata, "version: 0.1.0", "version: 'a b'", 1))),
"bad dep op": buildGem(t, []byte(strings.Replace(testMetadata, `- - "~>"`, `- - "=~"`, 1))),
"metadata bomb": buildGem(t, big),
"bad platform": buildGem(t, []byte(strings.Replace(testMetadata, "%PLATFORM%", "x/y", 1))),
"bad dependency": buildGem(t, []byte(strings.Replace(testMetadata, "name: json", "name: 'a|b'", 1))),
}
for name, gem := range cases {
if _, err := ParseGem(gem); err == nil {
t.Errorf("%s: accepted", name)
}
}
}
func TestVersionsLineMD5(t *testing.T) {
info := InfoBody([]string{"0.1.0 |checksum:aa", "0.2.0-java |checksum:bb"})
if string(info) != "---\n0.1.0 |checksum:aa\n0.2.0-java |checksum:bb\n" {
t.Fatalf("InfoBody = %q", info)
}
if got := VersionsLine("g", info); got != "g 0.1.0,0.2.0-java "+MD5Hex(info) {
t.Fatalf("VersionsLine = %q", got)
}
}
// Byte layouts below are taken from Ruby 3.3's Marshal.dump.
func TestRubyMarshalEncoding(t *testing.T) {
for n, want := range map[int]string{0: "\x00", 4: "\x09", 122: "\x7f", 123: "\x01\x7b", 255: "\x01\xff", 256: "\x02\x00\x01", 65536: "\x03\x00\x00\x01", -1: "\xfa", -123: "\x80"} {
var m rbMarshal
m.fixnum(n)
if m.String() != want {
t.Errorf("fixnum(%d) = %x, want %x", n, m.String(), want)
}
}
var m rbMarshal
m.str("3.5.22")
if m.String() != "I\"\x0b3.5.22\x06:\x06ET" {
t.Errorf("str = %q", m.String())
}
m.Reset()
m.version("0.1.0")
if m.String() != "U:\x11Gem::Version[\x06I\"\x0a0.1.0\x06:\x06ET" {
t.Errorf("version = %q", m.String())
}
m.Reset()
m.time(time.Date(2026, 10, 17, 0, 0, 0, 0, time.UTC))
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) {
s, err := ParseGem(testGem(t, "ruby"))
if err != nil {
t.Fatal(err)
}
q, err := QuickSpec(s)
if err != nil {
t.Fatal(err)
}
q2, _ := QuickSpec(s)
if !bytes.Equal(q, q2) {
t.Fatal("QuickSpec is not deterministic")
}
zr, err := zlib.NewReader(bytes.NewReader(q))
if err != nil {
t.Fatal(err)
}
raw, _ := io.ReadAll(zr)
head := "\x04\x08u:\x17Gem::Specification"
if !strings.HasPrefix(string(raw), head) {
t.Fatalf("quick spec prefix = %q", raw[:30])
}
inner := string(raw[len(head):])
at := strings.Index(inner, "\x04\x08[\x18")
var length rbMarshal
length.fixnum(len(inner) - at)
if at < 1 || inner[:at] != length.String() {
t.Fatalf("inner dump header = %q", inner[:8])
}
for _, want := range []string{"o:\x14Gem::Dependency", ":\x0cruntime", ":\x10development", "source_code_uri"} {
if !strings.Contains(string(raw), want) {
t.Errorf("quick spec missing %q", want)
}
}
}
+13 -12
View File
@@ -343,6 +343,9 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
return newEtag, false, err return newEtag, false, err
} }
// A transiently failed asset fails the scan so the ETag is not advanced and
// the retry re-derives it; an invalid package is skipped as no retry fixes it.
var failed []error
seen := map[string]bool{} seen := map[string]bool{}
for _, rel := range releases { for _, rel := range releases {
if rel.Draft { if rel.Draft {
@@ -373,10 +376,14 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
meta, err := p.deriveAsset(ctx, remote, asset, fp) meta, err := p.deriveAsset(ctx, remote, asset, fp)
if err != nil { if err != nil {
slog.Warn("github_rpm: derive asset failed", "remote", remote.Name, "asset", asset.Name, "error", err) slog.Warn("github_rpm: derive asset failed", "remote", remote.Name, "asset", asset.Name, "error", err)
if !errors.Is(err, provider.ErrInvalidPackage) {
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
}
continue continue
} }
if err := store.InsertRPMMetadata(ctx, meta); err != nil { if err := store.InsertRPMMetadata(ctx, meta); err != nil {
slog.Error("github_rpm: insert metadata failed", "remote", remote.Name, "asset", asset.Name, "error", err) slog.Error("github_rpm: insert metadata failed", "remote", remote.Name, "asset", asset.Name, "error", err)
failed = append(failed, fmt.Errorf("%s: %w", asset.Name, err))
continue continue
} }
slog.Info("github_rpm: derived asset", "remote", remote.Name, "name", meta.Name, "version", meta.Version, "arch", meta.Arch) slog.Info("github_rpm: derived asset", "remote", remote.Name, "name", meta.Name, "version", meta.Version, "arch", meta.Arch)
@@ -388,7 +395,7 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
_ = store.DeleteRPMMetadata(ctx, remote.Name, fp) _ = store.DeleteRPMMetadata(ctx, remote.Name, fp)
} }
} }
return newEtag, true, nil return newEtag, true, errors.Join(failed...)
} }
type ghRelease struct { type ghRelease struct {
@@ -569,12 +576,12 @@ func (p *GitHubProvider) fetchHeader(ctx context.Context, remote models.Remote,
} }
continue continue
} }
return nil, fmt.Errorf("parse rpm header: %w", perr) return nil, fmt.Errorf("%w: parse rpm header: %w", provider.ErrInvalidPackage, perr)
} }
} }
// rangeGet returns the first n bytes of downloadURL. full is true when the // rangeGet returns the first n bytes of downloadURL. full is true when the
// response body was shorter than n (i.e. we already have the whole object). // response holds the whole object.
func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, downloadURL string, n int64) ([]byte, bool, error) { func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, downloadURL string, n int64) ([]byte, bool, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil) req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
if err != nil { if err != nil {
@@ -596,15 +603,9 @@ func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, dow
} }
defer resp.Body.Close() defer resp.Body.Close()
if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent { if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
return nil, false, fmt.Errorf("range GET %s: status %d", downloadURL, resp.StatusCode) return nil, false, fmt.Errorf("range GET: %w", provider.AssetStatusError(downloadURL, resp))
} }
return provider.ReadPrefix(resp, n)
body, err := io.ReadAll(io.LimitReader(resp.Body, n))
if err != nil {
return nil, false, err
}
full := int64(len(body)) < n
return body, full, nil
} }
func (p *GitHubProvider) computeSHA256(ctx context.Context, remote models.Remote, downloadURL string) (string, error) { func (p *GitHubProvider) computeSHA256(ctx context.Context, remote models.Remote, downloadURL string) (string, error) {
@@ -627,7 +628,7 @@ func (p *GitHubProvider) computeSHA256(ctx context.Context, remote models.Remote
} }
defer resp.Body.Close() defer resp.Body.Close()
if resp.StatusCode != http.StatusOK { if resp.StatusCode != http.StatusOK {
return "", fmt.Errorf("GET %s: status %d", downloadURL, resp.StatusCode) return "", fmt.Errorf("GET: %w", provider.AssetStatusError(downloadURL, resp))
} }
h := sha256.New() h := sha256.New()
+34 -6
View File
@@ -1,11 +1,13 @@
package rpm package rpm
import ( import (
"cmp"
"compress/gzip" "compress/gzip"
"context" "context"
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"io" "io"
"net/http" "net/http"
@@ -24,15 +26,22 @@ import (
// fakeStore is an in-memory provider.RemoteMetadataStore keyed by file_path, // fakeStore is an in-memory provider.RemoteMetadataStore keyed by file_path,
// mirroring the (repo_name, file_path) uniqueness of the real table. // mirroring the (repo_name, file_path) uniqueness of the real table.
type fakeStore struct { type fakeStore struct {
mu sync.Mutex mu sync.Mutex
rows map[string]provider.RPMMetadata rows map[string]provider.RPMMetadata
insertFail map[string]int // package name -> inserts left to fail
} }
func newFakeStore() *fakeStore { return &fakeStore{rows: map[string]provider.RPMMetadata{}} } func newFakeStore() *fakeStore {
return &fakeStore{rows: map[string]provider.RPMMetadata{}, insertFail: map[string]int{}}
}
func (f *fakeStore) InsertRPMMetadata(_ context.Context, m *provider.RPMMetadata) error { func (f *fakeStore) InsertRPMMetadata(_ context.Context, m *provider.RPMMetadata) error {
f.mu.Lock() f.mu.Lock()
defer f.mu.Unlock() defer f.mu.Unlock()
if f.insertFail[m.Name] > 0 {
f.insertFail[m.Name]--
return errors.New("insert failed")
}
if _, ok := f.rows[m.FilePath]; ok { if _, ok := f.rows[m.FilePath]; ok {
return nil // ON CONFLICT DO NOTHING return nil // ON CONFLICT DO NOTHING
} }
@@ -79,15 +88,20 @@ type githubFixture struct {
releaseAuth string // Authorization header seen on the last releases request releaseAuth string // Authorization header seen on the last releases request
assetAuth string // Authorization header seen on the last asset request assetAuth string // Authorization header seen on the last asset request
failStatus int // when set, the releases list answers this status (rate-limit style) failStatus int // when set, the releases list answers this status (rate-limit style)
assetFail map[string]int // asset filename -> downloads left to answer failCode
failCode int // status for assetFail downloads; 0 = 500
shortRead map[string]int // asset filename -> ranged GETs left to close early
mu sync.Mutex mu sync.Mutex
} }
func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture { func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
t.Helper() t.Helper()
f := &githubFixture{ f := &githubFixture{
rpmBytes: map[string][]byte{}, rpmBytes: map[string][]byte{},
rangeHit: map[string]int{}, rangeHit: map[string]int{},
fullHit: map[string]int{}, fullHit: map[string]int{},
assetFail: map[string]int{},
shortRead: map[string]int{},
} }
f.rpmBytes["demo-1.2-3.x86_64.rpm"] = testsupport.MinimalRPM("demo", "1.2", "3", "x86_64") f.rpmBytes["demo-1.2-3.x86_64.rpm"] = testsupport.MinimalRPM("demo", "1.2", "3", "x86_64")
@@ -145,6 +159,17 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
} }
rng := r.Header.Get("Range") rng := r.Header.Get("Range")
f.mu.Lock() f.mu.Lock()
if f.assetFail[name] > 0 {
f.assetFail[name]--
code := cmp.Or(f.failCode, http.StatusInternalServerError)
f.mu.Unlock()
http.Error(w, "boom", code)
return
}
short := rng != "" && f.shortRead[name] > 0
if short {
f.shortRead[name]--
}
f.assetAuth = r.Header.Get("Authorization") f.assetAuth = r.Header.Get("Authorization")
if rng != "" { if rng != "" {
f.rangeHit[name]++ f.rangeHit[name]++
@@ -165,6 +190,9 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
end = len(body) - 1 end = len(body) - 1
} }
w.Header().Set("Content-Range", fmt.Sprintf("bytes 0-%d/%d", end, len(body))) w.Header().Set("Content-Range", fmt.Sprintf("bytes 0-%d/%d", end, len(body)))
if short {
end /= 2
}
w.Header().Set("Content-Length", strconv.Itoa(end+1)) w.Header().Set("Content-Length", strconv.Itoa(end+1))
w.WriteHeader(http.StatusPartialContent) w.WriteHeader(http.StatusPartialContent)
w.Write(body[:end+1]) w.Write(body[:end+1])
+71
View File
@@ -399,3 +399,74 @@ func TestSyncerFailedScanRetriesAfterBackoff(t *testing.T) {
t.Fatalf("recovery did not refresh metadata: %d rows", len(rows)) t.Fatalf("recovery did not refresh metadata: %d rows", len(rows))
} }
} }
// A transiently failed asset fails the scan: the ETag is not advanced, a retry
// is scheduled, and the retry against the unchanged release list derives it. An
// invalid package is skipped instead, since no retry can fix it.
func TestSyncerFailedAssetRetry(t *testing.T) {
for _, tc := range []struct {
name string
corrupt bool
failCode int
short bool
insert bool
wantRetry bool
}{
{name: "transient", wantRetry: true},
{name: "rate limited", failCode: http.StatusForbidden, wantRetry: true},
{name: "short read", short: true, wantRetry: true},
{name: "insert failure", insert: true, wantRetry: true},
{name: "invalid package", corrupt: true},
{name: "asset not found", failCode: http.StatusNotFound},
{name: "asset gone", failCode: http.StatusGone},
} {
t.Run(tc.name, func(t *testing.T) {
const other = "other-9-9.aarch64.rpm"
fx := newGitHubFixture(t, true)
fx.etag = `"v1"`
fx.rpmBytes[other] = testsupport.MinimalRPM("other", "9", "9", "aarch64")
store := newFakeSyncStore()
switch {
case tc.corrupt:
fx.rpmBytes[other] = []byte("not a package")
case tc.short:
fx.shortRead[other] = 1
case tc.insert:
store.insertFail["other"] = 1
case tc.failCode != 0 && !tc.wantRetry:
fx.failCode, fx.assetFail[other] = tc.failCode, 1<<30
default:
fx.failCode, fx.assetFail[other] = tc.failCode, 1
}
s := newSyncer(store, newTestProvider(), testSyncConfig())
remote := fx.remote()
bg := context.Background()
s.process(bg, syncJob{remote: remote, prime: true})
_, pending := store.retryAt[remote.Name]
if pending != tc.wantRetry {
t.Fatalf("retry pending=%v, want %v", pending, tc.wantRetry)
}
if _, saved := store.etags[remote.Name]; saved == tc.wantRetry {
t.Fatalf("etag saved=%v after first scan", saved)
}
if rows, _ := store.ListRPMMetadataEntries(bg, remote.Name); len(rows) != 1 {
t.Fatalf("healthy asset not served: %d rows", len(rows))
}
if !tc.wantRetry {
return
}
store.mu.Lock()
store.retryAt[remote.Name] = time.Now().Add(-time.Second)
store.mu.Unlock()
s.process(bg, syncJob{remote: remote})
if _, pending := store.retryAt[remote.Name]; pending || store.etags[remote.Name] != `"v1"` {
t.Fatalf("retry not recorded: pending=%v etag=%q", pending, store.etags[remote.Name])
}
if rows, _ := store.ListRPMMetadataEntries(bg, remote.Name); len(rows) != 2 {
t.Fatalf("retry did not derive the failed asset: %d rows", len(rows))
}
})
}
}
+45
View File
@@ -3,8 +3,10 @@ package provider
import ( import (
"errors" "errors"
"fmt" "fmt"
"io"
"net/http" "net/http"
"strconv" "strconv"
"strings"
"time" "time"
) )
@@ -43,6 +45,49 @@ func NewUpstreamStatusError(url string, resp *http.Response) *UpstreamStatusErro
return e return e
} }
// ErrInvalidPackage marks an asset whose bytes are not a valid package. It is
// permanent, so a scan skips the asset instead of failing and retrying.
var ErrInvalidPackage = errors.New("invalid package")
// AssetStatusError classifies a non-success asset download. A missing or
// unsatisfiable asset is permanent; anything else (403/429 rate limits, 5xx) is
// transient.
func AssetStatusError(url string, resp *http.Response) error {
switch resp.StatusCode {
case http.StatusNotFound, http.StatusGone, http.StatusRequestedRangeNotSatisfiable:
return fmt.Errorf("%w: %s: status %d", ErrInvalidPackage, url, resp.StatusCode)
}
return NewUpstreamStatusError(url, resp)
}
// ReadPrefix reads up to n bytes of a 200 or 206 asset response. full reports
// that the body is the whole object (per Content-Length or the Content-Range
// total, or a clean EOF before n when the total is unknown); a body ending before n bytes that is not the whole object is a
// transient short read.
func ReadPrefix(resp *http.Response, n int64) (body []byte, full bool, err error) {
body, err = io.ReadAll(io.LimitReader(resp.Body, n))
if err != nil {
return nil, false, err
}
total := resp.ContentLength
unknown := resp.StatusCode == http.StatusOK && total == -1
if resp.StatusCode == http.StatusPartialContent {
total = -1
if _, t, ok := strings.Cut(resp.Header.Get("Content-Range"), "/"); ok {
if v, perr := strconv.ParseInt(t, 10, 64); perr == nil {
total = v
}
unknown = t == "*"
}
}
// net/http errors on a premature close, so a clean EOF before n with an unknown total is the whole object.
full = total == int64(len(body)) || (unknown && int64(len(body)) < n)
if !full && int64(len(body)) < n {
return nil, false, fmt.Errorf("short read: got %d of %d bytes", len(body), n)
}
return body, full, nil
}
// SyncResult is a background scan's outcome, recorded when its sync lease is // SyncResult is a background scan's outcome, recorded when its sync lease is
// released. A failed scan keeps the prior sync time and ETag and schedules a // released. A failed scan keeps the prior sync time and ETag and schedules a
// retry after Backoff, doubled per consecutive failure up to MaxBackoff, and // retry after Backoff, doubled per consecutive failure up to MaxBackoff, and
+75
View File
@@ -3,7 +3,9 @@ package provider
import ( import (
"errors" "errors"
"fmt" "fmt"
"io"
"net/http" "net/http"
"net/http/httptest"
"strconv" "strconv"
"testing" "testing"
"time" "time"
@@ -61,3 +63,76 @@ func TestNewSyncResult(t *testing.T) {
t.Fatalf("wrapped hint not clamped to ttl: retry in %v", until) t.Fatalf("wrapped hint not clamped to ttl: retry in %v", until)
} }
} }
func TestAssetStatusErrorClassification(t *testing.T) {
for status, permanent := range map[int]bool{
http.StatusNotFound: true, http.StatusGone: true, http.StatusRequestedRangeNotSatisfiable: true,
http.StatusForbidden: false, http.StatusTooManyRequests: false, http.StatusInternalServerError: false, http.StatusBadGateway: false,
} {
err := AssetStatusError("u", &http.Response{StatusCode: status, Header: http.Header{}})
if errors.Is(err, ErrInvalidPackage) != permanent {
t.Errorf("status %d: permanent=%v, want %v", status, !permanent, permanent)
}
}
}
func TestReadPrefix(t *testing.T) {
cases := []struct {
name string
status int
cl int64
crange string
body string
n int64
wantFull bool
wantErr bool
}{
{name: "200 whole object", status: 200, cl: 5, body: "hello", n: 32, wantFull: true},
{name: "200 larger than n", status: 200, cl: 5, body: "hello", n: 4},
{name: "200 chunked shorter than n", status: 200, cl: -1, body: "hello", n: 32, wantFull: true},
{name: "200 chunked larger than n", status: 200, cl: -1, body: "hello", n: 4},
{name: "206 whole object", status: 206, crange: "bytes 0-4/5", body: "hello", n: 32, wantFull: true},
{name: "206 prefix", status: 206, crange: "bytes 0-3/100", body: "hell", n: 4},
{name: "206 short read", status: 206, crange: "bytes 0-31/100", body: "hello", n: 32, wantErr: true},
{name: "206 no content-range short", status: 206, body: "hello", n: 32, wantErr: true},
{name: "206 unknown total shorter than n", status: 206, cl: -1, crange: "bytes 0-4/*", body: "hello", n: 32, wantFull: true},
{name: "206 unknown total prefix", status: 206, cl: -1, crange: "bytes 0-3/*", body: "hell", n: 4},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
if tc.cl > 0 {
w.Header().Set("Content-Length", strconv.FormatInt(tc.cl, 10))
}
if tc.crange != "" {
w.Header().Set("Content-Range", tc.crange)
}
w.WriteHeader(tc.status)
_, _ = io.WriteString(w, tc.body)
w.(http.Flusher).Flush()
}))
defer srv.Close()
resp, err := http.Get(srv.URL)
if err != nil {
t.Fatal(err)
}
defer resp.Body.Close()
if tc.cl == -1 && resp.ContentLength != -1 {
t.Fatalf("response not chunked: ContentLength=%d", resp.ContentLength)
}
body, full, err := ReadPrefix(resp, tc.n)
if (err != nil) != tc.wantErr {
t.Fatalf("err=%v, wantErr=%v", err, tc.wantErr)
}
if err != nil {
if errors.Is(err, ErrInvalidPackage) {
t.Fatalf("short read must be transient: %v", err)
}
return
}
if full != tc.wantFull || int64(len(body)) > tc.n {
t.Fatalf("full=%v len=%d, want full=%v", full, len(body), tc.wantFull)
}
})
}
}
+26 -16
View File
@@ -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) 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 { if err != nil {
slog.Warn("lock acquire failed", "error", err) slog.Warn("lock acquire failed", "error", err)
} }
locked := token != ""
if !locked { if !locked {
// Another request holds the fetch lock. Poll the store until the leader // 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 { 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 { if class == ClassMutable && remote.CheckMutable {
@@ -228,10 +229,11 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error {
return fmt.Errorf("get remote: %w", err) return fmt.Errorf("get remote: %w", err)
} }
if !wildcard { 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 return err
} }
defer func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path) }() defer release()
if err := e.db.DeleteArtifact(ctx, remoteName, path); err != nil { if err := e.db.DeleteArtifact(ctx, remoteName, path); err != nil {
return fmt.Errorf("delete artifact: %w", err) return fmt.Errorf("delete artifact: %w", err)
} }
@@ -252,28 +254,36 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error {
return e.cache.ForgetPrefix(ctx, remoteName, prefix) return e.cache.ForgetPrefix(ctx, remoteName, prefix)
} }
// waitForLock takes the per-path fetch lock so an in-flight Fetch cannot // Lock takes the per-path Redis lock (the one Fetch holds) and returns its
// re-set TTL/ETag keys after an evict. It fails with a 503 when the lock // release. It fails with a 503 when the lock cannot be taken within wait or
// cannot be taken within evictLockWait or Redis errors. // Redis errors.
func (e *Engine) waitForLock(ctx context.Context, remoteName, path string) error { func (e *Engine) Lock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) (func(), error) {
deadline := time.Now().Add(e.evictLockWait) 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, token) }, nil
}
func (e *Engine) waitForLock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) (string, error) {
deadline := time.Now().Add(wait)
for { for {
ok, err := e.cache.AcquireLock(ctx, remoteName, path, fetchLockTTL) token, err := e.cache.AcquireLock(ctx, remoteName, path, ttl)
if ok { if token != "" {
return nil return token, nil
} }
if ctx.Err() != nil { if ctx.Err() != nil {
return ctx.Err() return "", ctx.Err()
} }
if err != nil { 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) { 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 { select {
case <-ctx.Done(): case <-ctx.Done():
return ctx.Err() return "", ctx.Err()
case <-time.After(50 * time.Millisecond): case <-time.After(50 * time.Millisecond):
} }
} }
+11 -9
View File
@@ -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() t.Helper()
ctx := context.Background() ctx := context.Background()
if ok, err := testCache.AcquireLock(ctx, remote, path, time.Minute); !ok || err != nil { token, err := testCache.AcquireLock(ctx, remote, path, time.Minute)
t.Fatalf("acquire: %v %v", ok, err) 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) { func TestEvictWaitsForFetchLock(t *testing.T) {
@@ -167,7 +169,7 @@ func TestEvictWaitsForFetchLock(t *testing.T) {
r := rpmRemote(t, "evict-lock", srv.URL) r := rpmRemote(t, "evict-lock", srv.URL)
ctx := context.Background() ctx := context.Background()
const path = "8/Everything/x86_64/repodata/repomd.xml" 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) done := make(chan error, 1)
go func() { done <- testEngine.Evict(ctx, r.Name, path) }() go func() { done <- testEngine.Evict(ctx, r.Name, path) }()
select { select {
@@ -175,13 +177,13 @@ func TestEvictWaitsForFetchLock(t *testing.T) {
t.Fatalf("evict returned while fetch lock held: %v", err) t.Fatalf("evict returned while fetch lock held: %v", err)
case <-time.After(200 * time.Millisecond): case <-time.After(200 * time.Millisecond):
} }
_ = testCache.ReleaseLock(ctx, r.Name, path) _ = testCache.ReleaseLock(ctx, r.Name, path, token)
if err := <-done; err != nil { if err := <-done; err != nil {
t.Fatalf("evict: %v", err) t.Fatalf("evict: %v", err)
} }
ok, err := testCache.AcquireLock(ctx, r.Name, path, time.Second) token, err := testCache.AcquireLock(ctx, r.Name, path, time.Second)
if !ok || err != nil { if token == "" || err != nil {
t.Fatalf("lock still held after evict: %v %v", ok, err) t.Fatalf("lock still held after evict: %q %v", token, err)
} }
} }
+1
View File
@@ -167,6 +167,7 @@ func (s *Server) routes() chi.Router {
proxyHandler := v1.NewProxyHandler(s.engine, s.virtEngine, s.db, s.store, s.localHandler) proxyHandler := v1.NewProxyHandler(s.engine, s.virtEngine, s.db, s.store, s.localHandler)
s.virtEngine.NPMLocal = proxyHandler.LocalNPMPackument s.virtEngine.NPMLocal = proxyHandler.LocalNPMPackument
s.virtEngine.GemsLocal = proxyHandler.LocalGemsIndex
r.Mount("/api/v1", proxyHandler.Routes()) r.Mount("/api/v1", proxyHandler.Routes())
r.Mount("/v2", proxyHandler.DockerV2Routes()) r.Mount("/v2", proxyHandler.DockerV2Routes())
+12
View File
@@ -32,6 +32,12 @@ type Engine struct {
// to the repo, or ErrNotFound. // to the repo, or ErrNotFound.
NPMLocal func(ctx context.Context, repo, pkg string) ([]byte, error) 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/<gem>, or
// ErrNotFound.
GemsLocal func(ctx context.Context, repo, path string) ([]byte, error)
mergeTTL time.Duration mergeTTL time.Duration
sf singleflight.Group 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 := &Engine{db: db, proxyEngine: proxyEngine, getRemote: db.GetRemote, mergeTTL: rpmMergeTTL}
e.rpmMember = e.fetchRPMMember e.rpmMember = e.fetchRPMMember
e.npmMember = e.fetchNPMMember e.npmMember = e.fetchNPMMember
e.gemsMember = e.fetchGemsMember
return e return e
} }
@@ -202,6 +209,11 @@ func (e *Engine) MemberRedirect(ctx context.Context, virt models.Virtual, path,
if !found || !slices.Contains(virt.Members, name) { if !found || !slices.Contains(virt.Members, name) {
return "", false, nil 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) segs, err := memberSegments(rest)
if err != nil { if err != nil {
return "", false, err return "", false, err
+294
View File
@@ -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/<gem>.
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/<full>.gem or
// quick/Marshal.4.8/<full>.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
}
+180
View File
@@ -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)
}
}
@@ -0,0 +1,12 @@
-- Sync state is dropped with its remote, so a recreated remote of the same name
-- starts with no inherited ETag or timing. NOT VALID skips checking rows left by
-- remotes deleted before this migration, which would otherwise fail it.
ALTER TABLE github_rpm_sync_state DROP CONSTRAINT IF EXISTS github_rpm_sync_state_remote_fk;
ALTER TABLE github_rpm_sync_state ADD CONSTRAINT github_rpm_sync_state_remote_fk
FOREIGN KEY (remote_name) REFERENCES remotes(name) ON DELETE CASCADE NOT VALID;
ALTER TABLE github_deb_sync_state DROP CONSTRAINT IF EXISTS github_deb_sync_state_remote_fk;
ALTER TABLE github_deb_sync_state ADD CONSTRAINT github_deb_sync_state_remote_fk
FOREIGN KEY (remote_name) REFERENCES remotes(name) ON DELETE CASCADE NOT VALID;
ALTER TABLE github_alpine_sync_state DROP CONSTRAINT IF EXISTS github_alpine_sync_state_remote_fk;
ALTER TABLE github_alpine_sync_state ADD CONSTRAINT github_alpine_sync_state_remote_fk
FOREIGN KEY (remote_name) REFERENCES remotes(name) ON DELETE CASCADE NOT VALID;
+20 -1
View File
@@ -2,7 +2,26 @@ import type { PackageTypeUI } from '../types';
const def: PackageTypeUI = { const def: PackageTypeUI = {
color: 'red', color: 'red',
snippets: ({ proxy }) => { snippets: ({ name, url, proxy, isLocal }) => {
if (isLocal) {
const src = `${url}/api/v1/local/${name}`;
return [
{
title: 'Push and yank',
language: 'bash',
code: `GEM_HOST_API_KEY=unused gem push --host ${src} <gem>-<version>.gem
GEM_HOST_API_KEY=unused gem yank <gem> -v <version> --host ${src}`,
note: 'gem push needs an API key; any value works. Pushing an existing version is rejected.',
},
{
title: 'Install',
language: 'bash',
code: `gem install <gem> --source ${src}/
# Gemfile
source "${src}/"`,
},
];
}
return [ return [
{ {
title: 'Use this remote as the RubyGems source', title: 'Use this remote as the RubyGems source',