diff --git a/internal/provider/alpine/github.go b/internal/provider/alpine/github.go index 9461166..46aad38 100644 --- a/internal/provider/alpine/github.go +++ b/internal/provider/alpine/github.go @@ -601,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 { @@ -623,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 diff --git a/internal/provider/alpine/github_test.go b/internal/provider/alpine/github_test.go index 53dd777..69f27f7 100644 --- a/internal/provider/alpine/github_test.go +++ b/internal/provider/alpine/github_test.go @@ -8,6 +8,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -503,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) + } +} diff --git a/internal/provider/deb/github.go b/internal/provider/deb/github.go index aa2305a..83c85de 100644 --- a/internal/provider/deb/github.go +++ b/internal/provider/deb/github.go @@ -574,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 { @@ -596,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) { @@ -627,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() diff --git a/internal/provider/deb/github_test.go b/internal/provider/deb/github_test.go index 519a8c8..b03956d 100644 --- a/internal/provider/deb/github_test.go +++ b/internal/provider/deb/github_test.go @@ -7,6 +7,7 @@ import ( "crypto/sha256" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "net/http" @@ -468,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) + } +} diff --git a/internal/provider/rpm/github.go b/internal/provider/rpm/github.go index e48d4f0..707b54d 100644 --- a/internal/provider/rpm/github.go +++ b/internal/provider/rpm/github.go @@ -581,7 +581,7 @@ func (p *GitHubProvider) fetchHeader(ctx context.Context, remote models.Remote, } // 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 { @@ -603,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) { @@ -634,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() diff --git a/internal/provider/rpm/github_test.go b/internal/provider/rpm/github_test.go index fccb666..6298c4d 100644 --- a/internal/provider/rpm/github_test.go +++ b/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,7 +88,9 @@ 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 500 + 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 } @@ -90,6 +101,7 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture { 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") @@ -149,10 +161,15 @@ func newGitHubFixture(t *testing.T, withDigest bool) *githubFixture { 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", http.StatusInternalServerError) + 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]++ @@ -173,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]) diff --git a/internal/provider/rpm/syncer_test.go b/internal/provider/rpm/syncer_test.go index a01c785..be6d9c9 100644 --- a/internal/provider/rpm/syncer_test.go +++ b/internal/provider/rpm/syncer_test.go @@ -407,21 +407,37 @@ 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"` - if tc.corrupt { - fx.rpmBytes["other-9-9.aarch64.rpm"] = []byte("not a package") - } else { - fx.rpmBytes["other-9-9.aarch64.rpm"] = testsupport.MinimalRPM("other", "9", "9", "aarch64") - fx.assetFail["other-9-9.aarch64.rpm"] = 1 - } + 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() diff --git a/internal/provider/syncretry.go b/internal/provider/syncretry.go index 27dd4ba..cec262e 100644 --- a/internal/provider/syncretry.go +++ b/internal/provider/syncretry.go @@ -3,8 +3,10 @@ package provider import ( "errors" "fmt" + "io" "net/http" "strconv" + "strings" "time" ) @@ -47,6 +49,42 @@ func NewUpstreamStatusError(url string, resp *http.Response) *UpstreamStatusErro // 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); 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 + 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 + } + } + } + full = total == int64(len(body)) + 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 diff --git a/internal/provider/syncretry_test.go b/internal/provider/syncretry_test.go index 289a1d7..4feab63 100644 --- a/internal/provider/syncretry_test.go +++ b/internal/provider/syncretry_test.go @@ -3,8 +3,10 @@ package provider import ( "errors" "fmt" + "io" "net/http" "strconv" + "strings" "testing" "time" ) @@ -61,3 +63,57 @@ 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: 100, body: "hello", n: 4}, + {name: "200 unknown length short", status: 200, cl: -1, body: "hello", n: 32, wantErr: true}, + {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}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + resp := &http.Response{StatusCode: tc.status, ContentLength: tc.cl, Header: http.Header{}, Body: io.NopCloser(strings.NewReader(tc.body))} + if tc.crange != "" { + resp.Header.Set("Content-Range", tc.crange) + } + 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) + } + }) + } +}