Merge master into benvin/remote-query-forwarder

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>

internal/api/v1/forwardquery_test.go

@@ -0,0 +1,58 @@
+package v1
+
+import (
+	"net/url"
+	"strings"
+	"testing"
+
+	"git.unkin.net/unkin/artifactapi/internal/provider"
+)
+
+type plainProvider struct{ provider.Provider }
+
+type forwardingProvider struct{ plainProvider }
+
+func digits(v string) (string, bool) {
+	if v == "" || strings.Trim(v, "0123456789") != "" {
+		return "", false
+	}
+	return strings.TrimLeft(v, "0"), true
+}
+
+func (forwardingProvider) ForwardedQuery(path string) map[string]provider.QueryValidator {
+	if strings.HasSuffix(path, ".tar.gz") {
+		return nil
+	}
+	return map[string]provider.QueryValidator{"limit": digits, "offset": digits}
+}
+
+func TestForwardQuery(t *testing.T) {
+	tests := []struct {
+		name  string
+		prov  provider.Provider
+		path  string
+		query string
+		want  string
+	}{
+		{"non-forwarder keeps path", plainProvider{}, "a/versions/", "limit=10&offset=5", "a/versions/"},
+		{"no query", forwardingProvider{}, "a/versions/", "", "a/versions/"},
+		{"allowlisted params sorted", forwardingProvider{}, "a/versions/", "offset=5&limit=10", "a/versions/?limit=10&offset=5"},
+		{"unknown params dropped", forwardingProvider{}, "a/versions/", "x=1&limit=10&cb=zz", "a/versions/?limit=10"},
+		{"only unknown params", forwardingProvider{}, "a/versions/", "x=1", "a/versions/"},
+		{"first value wins", forwardingProvider{}, "a/versions/", "limit=10&limit=20", "a/versions/?limit=10"},
+		{"invalid value dropped", forwardingProvider{}, "a/versions/", "limit=1%202&offset=5", "a/versions/?offset=5"},
+		{"value canonicalised", forwardingProvider{}, "a/versions/", "limit=0010", "a/versions/?limit=10"},
+		{"path without allowlist", forwardingProvider{}, "a/b-1.0.0.tar.gz", "limit=10", "a/b-1.0.0.tar.gz"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			q, err := url.ParseQuery(tt.query)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if got := forwardQuery(tt.prov, tt.path, q); got != tt.want {
+				t.Errorf("forwardQuery = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}

internal/api/v1/gems_handler_test.go

@@ -231,8 +231,9 @@ func TestGemsLockTimeoutIs503(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if ok, err := redis.AcquireLock(context.Background(), repo, "_gemlock/busy-1.0.0", time.Minute); !ok || err != nil {
-		t.Fatalf("hold lock = %v %v", ok, err)
+	token, err := redis.AcquireLock(context.Background(), repo, "_gemlock/busy-1.0.0", time.Minute)
+	if token == "" || err != nil {
+		t.Fatalf("hold lock = %q %v", token, err)
 	}
 	prev := gemsLockWait
 	gemsLockWait = 100 * time.Millisecond
@@ -244,7 +245,7 @@ func TestGemsLockTimeoutIs503(t *testing.T) {
 	if code, _ := npmDo(t, http.MethodDelete, base+"api/v1/gems/yank?gem_name=busy&version=1.0.0", ""); code != http.StatusServiceUnavailable {
 		t.Fatalf("yank under held lock = %d, want 503", code)
 	}
-	if err := redis.ReleaseLock(context.Background(), repo, "_gemlock/busy-1.0.0"); err != nil {
+	if err := redis.ReleaseLock(context.Background(), repo, "_gemlock/busy-1.0.0", token); err != nil {
 		t.Fatal(err)
 	}
 	if code, body := npmDo(t, http.MethodPost, base+"api/v1/gems", string(testGemFile(t, "busy", "1.0.0", "ruby"))); code != http.StatusOK {

internal/api/v1/proxy.go

@@ -8,6 +8,7 @@ import (
 	"log/slog"
 	"net"
 	"net/http"
+	"net/url"
 	"strconv"
 	"strings"
 
@@ -115,7 +116,7 @@ func (h *ProxyHandler) handleProxy(w http.ResponseWriter, r *http.Request) {
 }
 
 func (h *ProxyHandler) serveFetch(w http.ResponseWriter, r *http.Request, remote *models.Remote, prov provider.Provider, path string) {
-	result, err := h.engine.Fetch(r.Context(), *remote, path, prov, r.Header)
+	result, err := h.engine.Fetch(r.Context(), *remote, forwardQuery(prov, path, r.URL.Query()), prov, r.Header)
 	if err != nil {
 		var proxyErr *proxy.ProxyError
 		if errors.As(err, &proxyErr) {
@@ -325,6 +326,25 @@ func Scheme(r *http.Request) string {
 	return "http"
 }
 
+// forwardQuery appends the provider's validated query params to path in a
+// canonical (sorted, single-valued) form, since the result is the cache key.
+func forwardQuery(prov provider.Provider, path string, query url.Values) string {
+	f, ok := prov.(provider.QueryForwarder)
+	if !ok {
+		return path
+	}
+	kept := url.Values{}
+	for name, valid := range f.ForwardedQuery(path) {
+		if v, ok := valid(query.Get(name)); ok {
+			kept.Set(name, v)
+		}
+	}
+	if len(kept) == 0 {
+		return path
+	}
+	return path + "?" + kept.Encode()
+}
+
 // rewriteServed rewrites a cached body for the requesting host and marks the
 // response uncacheable by shared caches, since it varies per Host.
 func rewriteServed(w http.ResponseWriter, r *http.Request, sr provider.ServeRewriter, remote models.Remote, src io.Reader) ([]byte, bool) {

internal/cache/cache_test.go

@@ -61,18 +61,54 @@ func TestTTL(t *testing.T) {
 func TestLock(t *testing.T) {
 	requireRedis(t)
 	ctx := context.Background()
-	ok, err := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute)
-	if err != nil || !ok {
-		t.Fatalf("first acquire should succeed: %v %v", ok, err)
+	token, err := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute)
+	if err != nil || token == "" {
+		t.Fatalf("first acquire should succeed: %q %v", token, err)
 	}
-	if ok, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); ok {
+	if again, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); again != "" {
 		t.Error("second acquire should fail while held")
 	}
-	if err := testRedis.ReleaseLock(ctx, "r", "lockpath"); err != nil {
+	if err := testRedis.ReleaseLock(ctx, "r", "lockpath", "not-the-owner"); err != nil {
 		t.Fatal(err)
 	}
-	if ok, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); !ok {
-		t.Error("acquire should succeed after release")
+	if again, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); again != "" {
+		t.Error("release with a foreign token must not drop the lock")
+	}
+	if err := testRedis.ReleaseLock(ctx, "r", "lockpath", token); err != nil {
+		t.Fatal(err)
+	}
+	next, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute)
+	if next == "" {
+		t.Fatal("acquire should succeed after release")
+	}
+	if next == token {
+		t.Error("tokens must differ between holders")
+	}
+}
+
+func TestStaleReleaseKeepsNextHolder(t *testing.T) {
+	requireRedis(t)
+	ctx := context.Background()
+	stale, err := testRedis.AcquireLock(ctx, "r", "stale", 50*time.Millisecond)
+	if err != nil || stale == "" {
+		t.Fatalf("acquire: %q %v", stale, err)
+	}
+	time.Sleep(100 * time.Millisecond)
+	fresh, err := testRedis.AcquireLock(ctx, "r", "stale", time.Minute)
+	if err != nil || fresh == "" {
+		t.Fatalf("acquire after expiry: %q %v", fresh, err)
+	}
+	if err := testRedis.ReleaseLock(ctx, "r", "stale", stale); err != nil {
+		t.Fatal(err)
+	}
+	if again, _ := testRedis.AcquireLock(ctx, "r", "stale", time.Minute); again != "" {
+		t.Fatal("stale holder's release dropped the second holder's lock")
+	}
+	if err := testRedis.ReleaseLock(ctx, "r", "stale", fresh); err != nil {
+		t.Fatal(err)
+	}
+	if again, _ := testRedis.AcquireLock(ctx, "r", "stale", time.Minute); again == "" {
+		t.Error("owner release should free the lock")
 	}
 }
 

internal/cache/redis.go

@@ -2,6 +2,7 @@ package cache
 
 import (
 	"context"
+	"crypto/rand"
 	"fmt"
 	"strings"
 	"time"
@@ -46,15 +47,23 @@ func (r *Redis) CheckTTL(ctx context.Context, remote, path string) (bool, error)
 	return exists > 0, nil
 }
 
-func (r *Redis) AcquireLock(ctx context.Context, remote, path string, ttl time.Duration) (bool, error) {
+// AcquireLock returns the owner token on success and "" when the lock is held.
+func (r *Redis) AcquireLock(ctx context.Context, remote, path string, ttl time.Duration) (string, error) {
 	key := fmt.Sprintf("lock:%s:%s", remote, path)
-	ok, err := r.client.SetNX(ctx, key, "1", ttl).Result()
-	return ok, err
+	token := rand.Text()
+	ok, err := r.client.SetNX(ctx, key, token, ttl).Result()
+	if err != nil || !ok {
+		return "", err
+	}
+	return token, nil
 }
 
-func (r *Redis) ReleaseLock(ctx context.Context, remote, path string) error {
+var releaseScript = redis.NewScript(`if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end`)
+
+// ReleaseLock deletes the lock only while token still owns it.
+func (r *Redis) ReleaseLock(ctx context.Context, remote, path, token string) error {
 	key := fmt.Sprintf("lock:%s:%s", remote, path)
-	return r.client.Del(ctx, key).Err()
+	return releaseScript.Run(ctx, r.client, []string{key}, token).Err()
 }
 
 // SetETag stores etag for path, or drops the stored one when etag is empty.

internal/database/github_sync_test.go

@@ -207,3 +207,72 @@ func TestGitHubSyncLeaseFailureBackoff(t *testing.T) {
 		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)
+			}
+		})
+	}
+}

internal/provider/alpine/github.go

@@ -372,6 +372,9 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
 		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{}
 	for _, rel := range releases {
 		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)
 			if err != nil {
 				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
 			}
 			if err := inserter.InsertAlpineMetadata(ctx, meta); err != nil {
 				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
 			}
 			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)
 		}
 	}
-	return newEtag, true, nil
+	return newEtag, true, errors.Join(failed...)
 }
 
 type ghRelease struct {
@@ -496,7 +503,7 @@ func (p *GitHubProvider) deriveAsset(ctx context.Context, remote models.Remote,
 		return nil, err
 	}
 	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
@@ -526,13 +533,13 @@ func (p *GitHubProvider) fetchPkginfo(ctx context.Context, remote models.Remote,
 		}
 		meta, complete, perr := pkginfoFromPrefix(body)
 		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 {
 			return meta, nil
 		}
 		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
 		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
-// 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) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
 	if err != nil {
@@ -616,15 +623,9 @@ func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, dow
 	}
 	defer resp.Body.Close()
 	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))
 	}
-
-	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
+	return provider.ReadPrefix(resp, n)
 }
 
 // assetPath is the package's location relative to github.com — the path the

internal/provider/alpine/github_test.go

@@ -8,6 +8,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -88,15 +89,17 @@ type githubFixture struct {
 	notModHit   int
 	releaseAuth string
 	assetAuth   string
+	assetFail   map[string]int // asset filename -> downloads left to answer 500
 	mu          sync.Mutex
 }
 
 func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
 	t.Helper()
 	f := &githubFixture{
-		apkBytes: map[string][]byte{},
-		rangeHit: map[string]int{},
-		fullHit:  map[string]int{},
+		apkBytes:  map[string][]byte{},
+		rangeHit:  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")
 
@@ -146,6 +149,12 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
 		}
 		rng := r.Header.Get("Range")
 		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")
 		if rng != "" {
 			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)
+	}
+}

internal/provider/alpine/syncer_test.go

@@ -308,3 +308,58 @@ func TestSyncerPrimeBypassesRecencyPeriodicDoesNot(t *testing.T) {
 		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))
+			}
+		})
+	}
+}

internal/provider/cap_query_forwarder.go

@@ -0,0 +1,12 @@
+package provider
+
+// QueryForwarder is a remote provider whose upstream answers depend on the
+// query string (e.g. paginated JSON APIs). Only the named params for path are
+// forwarded upstream and kept in the cache key, each as the canonical value
+// its validator returns; a value the validator rejects is dropped.
+type QueryForwarder interface {
+	ForwardedQuery(path string) map[string]QueryValidator
+}
+
+// QueryValidator canonicalises a raw query value, reporting false to drop it.
+type QueryValidator func(v string) (string, bool)

internal/provider/deb/github.go

@@ -333,6 +333,9 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
 		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{}
 	for _, rel := range releases {
 		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)
 			if err != nil {
 				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
 			}
 			if err := store.InsertDebMetadata(ctx, meta); err != nil {
 				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
 			}
 			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)
 		}
 	}
-	return newEtag, true, nil
+	return newEtag, true, errors.Join(failed...)
 }
 
 type ghRelease struct {
@@ -468,7 +475,7 @@ func (p *GitHubProvider) deriveAsset(ctx context.Context, remote models.Remote,
 		Size:         asset.Size,
 	}
 	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
@@ -503,13 +510,13 @@ func (p *GitHubProvider) fetchControl(ctx context.Context, remote models.Remote,
 		}
 		control, complete, perr := controlFromPrefix(body)
 		if perr != nil {
-			return "", fmt.Errorf("parse deb control: %w", perr)
+			return "", fmt.Errorf("%w: parse deb control: %w", provider.ErrInvalidPackage, perr)
 		}
 		if complete {
 			return control, nil
 		}
 		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
 		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
-// 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) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
 	if err != nil {
@@ -589,15 +596,9 @@ func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, dow
 	}
 	defer resp.Body.Close()
 	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))
 	}
-
-	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
+	return provider.ReadPrefix(resp, n)
 }
 
 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()
 	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()

internal/provider/deb/github_test.go

@@ -7,6 +7,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -81,15 +82,17 @@ type githubFixture struct {
 	notModHit   int
 	releaseAuth string
 	assetAuth   string
+	assetFail   map[string]int // asset filename -> downloads left to answer 500
 	mu          sync.Mutex
 }
 
 func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
 	t.Helper()
 	f := &githubFixture{
-		debBytes: map[string][]byte{},
-		rangeHit: map[string]int{},
-		fullHit:  map[string]int{},
+		debBytes:  map[string][]byte{},
+		rangeHit:  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")
 
@@ -139,6 +142,12 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
 		}
 		rng := r.Header.Get("Range")
 		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")
 		if rng != "" {
 			f.rangeHit[name]++
@@ -460,3 +469,13 @@ func TestGitHubAssetPatternFilter(t *testing.T) {
 		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)
+	}
+}

internal/provider/deb/syncer_test.go

@@ -308,3 +308,58 @@ func TestSyncerPrimeBypassesRecencyPeriodicDoesNot(t *testing.T) {
 		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))
+			}
+		})
+	}
+}

internal/provider/rpm/github.go

@@ -343,6 +343,9 @@ func (p *GitHubProvider) scanWithState(ctx context.Context, remote models.Remote
 		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{}
 	for _, rel := range releases {
 		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)
 			if err != nil {
 				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
 			}
 			if err := store.InsertRPMMetadata(ctx, meta); err != nil {
 				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
 			}
 			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)
 		}
 	}
-	return newEtag, true, nil
+	return newEtag, true, errors.Join(failed...)
 }
 
 type ghRelease struct {
@@ -569,12 +576,12 @@ func (p *GitHubProvider) fetchHeader(ctx context.Context, remote models.Remote,
 			}
 			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
-// 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) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, downloadURL, nil)
 	if err != nil {
@@ -596,15 +603,9 @@ func (p *GitHubProvider) rangeGet(ctx context.Context, remote models.Remote, dow
 	}
 	defer resp.Body.Close()
 	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))
 	}
-
-	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
+	return provider.ReadPrefix(resp, n)
 }
 
 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()
 	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()

internal/provider/rpm/github_test.go

@@ -1,11 +1,13 @@
 package rpm
 
 import (
+	"cmp"
 	"compress/gzip"
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -24,15 +26,22 @@ import (
 // fakeStore is an in-memory provider.RemoteMetadataStore keyed by file_path,
 // mirroring the (repo_name, file_path) uniqueness of the real table.
 type fakeStore struct {
-	mu   sync.Mutex
-	rows map[string]provider.RPMMetadata
+	mu         sync.Mutex
+	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 {
 	f.mu.Lock()
 	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 {
 		return nil // ON CONFLICT DO NOTHING
 	}
@@ -79,15 +88,20 @@ type githubFixture struct {
 	releaseAuth string            // Authorization header seen on the last releases request
 	assetAuth   string            // Authorization header seen on the last asset request
 	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
 }
 
 func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
 	t.Helper()
 	f := &githubFixture{
-		rpmBytes: map[string][]byte{},
-		rangeHit: map[string]int{},
-		fullHit:  map[string]int{},
+		rpmBytes:  map[string][]byte{},
+		rangeHit:  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")
 
@@ -145,6 +159,17 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
 		}
 		rng := r.Header.Get("Range")
 		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")
 		if rng != "" {
 			f.rangeHit[name]++
@@ -165,6 +190,9 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture {
 			end = len(body) - 1
 		}
 		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.WriteHeader(http.StatusPartialContent)
 		w.Write(body[:end+1])

internal/provider/rpm/syncer_test.go

@@ -399,3 +399,74 @@ func TestSyncerFailedScanRetriesAfterBackoff(t *testing.T) {
 		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))
+			}
+		})
+	}
+}

internal/provider/syncretry.go

@@ -3,8 +3,10 @@ package provider
 import (
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"strconv"
+	"strings"
 	"time"
 )
 
@@ -43,6 +45,49 @@ func NewUpstreamStatusError(url string, resp *http.Response) *UpstreamStatusErro
 	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
 // 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

internal/provider/syncretry_test.go

@@ -3,7 +3,9 @@ package provider
 import (
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
+	"net/http/httptest"
 	"strconv"
 	"testing"
 	"time"
@@ -61,3 +63,76 @@ func TestNewSyncResult(t *testing.T) {
 		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)
+			}
+		})
+	}
+}

internal/proxy/classifier.go

@@ -2,6 +2,7 @@ package proxy
 
 import (
 	"regexp"
+	"strings"
 	"sync"
 
 	"git.unkin.net/unkin/artifactapi/internal/provider"
@@ -37,7 +38,9 @@ func NewClassifier(p provider.Provider) *Classifier {
 	return &Classifier{provider: p}
 }
 
+// Classify judges the bare path; a forwarded query must not change the verdict.
 func (c *Classifier) Classify(remote models.Remote, path string) Classification {
+	path, _, _ = strings.Cut(path, "?")
 	if matchesAny(path, compilePatterns(remote.Blocklist)) {
 		return ClassDenied
 	}

internal/proxy/classifier_extra_test.go

@@ -38,6 +38,24 @@ func TestClassifierBranches(t *testing.T) {
 	}
 }
 
+func TestClassifierIgnoresForwardedQuery(t *testing.T) {
+	gp, err := provider.Get(models.PackageGeneric)
+	if err != nil {
+		t.Fatal(err)
+	}
+	c := NewClassifier(gp)
+	if got := c.Classify(models.Remote{Blocklist: []string{`\.exe$`}}, "x.exe?limit=1"); got != ClassDenied {
+		t.Errorf("blocklist bypassed by query: %v", got)
+	}
+	allow := models.Remote{Patterns: []string{`/versions/$`}}
+	if got := c.Classify(allow, "a/versions/?limit=1&offset=100"); got != ClassImmutable {
+		t.Errorf("allowlisted paginated path = %v, want immutable", got)
+	}
+	if got := c.Classify(models.Remote{MutablePatterns: []string{`/versions/$`}}, "a/versions/?offset=100"); got != ClassMutable {
+		t.Errorf("mutable pattern on paginated path = %v, want mutable", got)
+	}
+}
+
 func TestClassificationString(t *testing.T) {
 	for c, want := range map[Classification]string{
 		ClassImmutable:     "immutable",

internal/proxy/engine.go

@@ -134,10 +134,11 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p
 		slog.Warn("cache hit but S3 miss, re-fetching", "remote", remote.Name, "path", path)
 	}
 
-	locked, err := e.cache.AcquireLock(ctx, remote.Name, path, fetchLockTTL)
+	token, err := e.cache.AcquireLock(ctx, remote.Name, path, fetchLockTTL)
 	if err != nil {
 		slog.Warn("lock acquire failed", "error", err)
 	}
+	locked := token != ""
 
 	if !locked {
 		// Another request holds the fetch lock. Poll the store until the leader
@@ -151,7 +152,7 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p
 	}
 
 	if locked {
-		defer e.cache.ReleaseLock(ctx, remote.Name, path)
+		defer func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remote.Name, path, token) }()
 	}
 
 	if class == ClassMutable && remote.CheckMutable {
@@ -216,7 +217,8 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p
 
 // Evict drops path from every cache layer (artifact row, index object, Redis
 // freshness and ETag keys) so the next request refetches from upstream. A
-// trailing "/*" evicts every path under that directory.
+// trailing "/*" evicts every path under that directory; a plain path also
+// drops its forwarded-query variants ("path?...").
 func (e *Engine) Evict(ctx context.Context, remoteName, path string) error {
 	prefix, wildcard := strings.CutSuffix(path, "*")
 	if wildcard && !strings.HasSuffix(prefix, "/") {
@@ -239,11 +241,23 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error {
 		if err := e.store.Delete(ctx, storage.IndexKey(remoteName, path)); err != nil {
 			return fmt.Errorf("delete index: %w", err)
 		}
-		return e.cache.ForgetPath(ctx, remoteName, path)
+		if err := e.cache.ForgetPath(ctx, remoteName, path); err != nil {
+			return err
+		}
+		if strings.Contains(path, "?") {
+			return nil
+		}
+		return e.evictPrefix(ctx, remoteName, path+"?")
 	}
 	// ponytail: no lock for wildcards; a Fetch already in flight under the
 	// prefix can re-cache its path after the evict. Per-path locks over a
 	// directory would close it if that ever matters.
+	return e.evictPrefix(ctx, remoteName, prefix)
+}
+
+// evictPrefix drops every cached path starting with prefix, including the
+// forwarded-query variants ("path?...") of a single path.
+func (e *Engine) evictPrefix(ctx context.Context, remoteName, prefix string) error {
 	if err := e.db.DeleteArtifactsByPrefix(ctx, remoteName, prefix); err != nil {
 		return fmt.Errorf("delete artifacts: %w", err)
 	}
@@ -257,31 +271,32 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error {
 // release. It fails with a 503 when the lock cannot be taken within wait or
 // Redis errors.
 func (e *Engine) Lock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) (func(), error) {
-	if err := e.waitForLock(ctx, remoteName, path, wait, ttl); err != nil {
+	token, err := e.waitForLock(ctx, remoteName, path, wait, ttl)
+	if err != nil {
 		return nil, err
 	}
-	return func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path) }, nil
+	return func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path, token) }, nil
 }
 
-func (e *Engine) waitForLock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) error {
+func (e *Engine) waitForLock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) (string, error) {
 	deadline := time.Now().Add(wait)
 	for {
-		ok, err := e.cache.AcquireLock(ctx, remoteName, path, ttl)
-		if ok {
-			return nil
+		token, err := e.cache.AcquireLock(ctx, remoteName, path, ttl)
+		if token != "" {
+			return token, nil
 		}
 		if ctx.Err() != nil {
-			return ctx.Err()
+			return "", ctx.Err()
 		}
 		if err != nil {
-			return &ProxyError{Status: http.StatusServiceUnavailable, Message: fmt.Sprintf("fetch lock: %v", err)}
+			return "", &ProxyError{Status: http.StatusServiceUnavailable, Message: fmt.Sprintf("lock: %v", err)}
 		}
 		if time.Now().After(deadline) {
-			return &ProxyError{Status: http.StatusServiceUnavailable, Message: "fetch in progress, retry evict"}
+			return "", &ProxyError{Status: http.StatusServiceUnavailable, Message: "resource busy, retry later"}
 		}
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
+			return "", ctx.Err()
 		case <-time.After(50 * time.Millisecond):
 		}
 	}

internal/proxy/evict_test.go

@@ -152,13 +152,15 @@ func TestEvictUnknownRemote(t *testing.T) {
 	}
 }
 
-func holdLock(t *testing.T, remote, path string) {
+func holdLock(t *testing.T, remote, path string) string {
 	t.Helper()
 	ctx := context.Background()
-	if ok, err := testCache.AcquireLock(ctx, remote, path, time.Minute); !ok || err != nil {
-		t.Fatalf("acquire: %v %v", ok, err)
+	token, err := testCache.AcquireLock(ctx, remote, path, time.Minute)
+	if token == "" || err != nil {
+		t.Fatalf("acquire: %q %v", token, err)
 	}
-	t.Cleanup(func() { _ = testCache.ReleaseLock(ctx, remote, path) })
+	t.Cleanup(func() { _ = testCache.ReleaseLock(ctx, remote, path, token) })
+	return token
 }
 
 func TestEvictWaitsForFetchLock(t *testing.T) {
@@ -167,7 +169,7 @@ func TestEvictWaitsForFetchLock(t *testing.T) {
 	r := rpmRemote(t, "evict-lock", srv.URL)
 	ctx := context.Background()
 	const path = "8/Everything/x86_64/repodata/repomd.xml"
-	holdLock(t, r.Name, path)
+	token := holdLock(t, r.Name, path)
 	done := make(chan error, 1)
 	go func() { done <- testEngine.Evict(ctx, r.Name, path) }()
 	select {
@@ -175,13 +177,13 @@ func TestEvictWaitsForFetchLock(t *testing.T) {
 		t.Fatalf("evict returned while fetch lock held: %v", err)
 	case <-time.After(200 * time.Millisecond):
 	}
-	_ = testCache.ReleaseLock(ctx, r.Name, path)
+	_ = testCache.ReleaseLock(ctx, r.Name, path, token)
 	if err := <-done; err != nil {
 		t.Fatalf("evict: %v", err)
 	}
-	ok, err := testCache.AcquireLock(ctx, r.Name, path, time.Second)
-	if !ok || err != nil {
-		t.Fatalf("lock still held after evict: %v %v", ok, err)
+	token, err := testCache.AcquireLock(ctx, r.Name, path, time.Second)
+	if token == "" || err != nil {
+		t.Fatalf("lock still held after evict: %q %v", token, err)
 	}
 }
 
@@ -269,3 +271,28 @@ func TestEvictWildcardClearsPrefixOnly(t *testing.T) {
 		t.Errorf("path outside prefix = %q, want cached rev1", got)
 	}
 }
+
+func TestEvictPathDropsQueryVariants(t *testing.T) {
+	requireStack(t)
+	srv, _, _ := changingUpstream(t)
+	r := rpmRemote(t, "evict-query", srv.URL)
+	const (
+		path    = "8/Everything/x86_64/repodata/repomd.xml"
+		variant = path + "?limit=10"
+		sibling = "8/Everything/x86_64/repodata/repomd.xml.asc"
+	)
+	for _, p := range []string{path, variant, sibling} {
+		fetchBody(t, r, p)
+	}
+	if err := testEngine.Evict(context.Background(), r.Name, path); err != nil {
+		t.Fatalf("evict: %v", err)
+	}
+	for _, p := range []string{path, variant} {
+		if c := layers(t, r.Name, p); c != (cached{}) {
+			t.Errorf("%s after evict = %+v, want every layer gone", p, c)
+		}
+	}
+	if c := layers(t, r.Name, sibling); c == (cached{}) {
+		t.Errorf("%s sharing the name prefix was evicted", sibling)
+	}
+}

migrations/0003_github_sync_state_cascade.sql

@@ -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;