Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| e59d9728b2 | |||
| e4833b6f96 | |||
| 7f65fc377c | |||
| 00edeb478b | |||
| f5d30cc96e | |||
| a4a4fe57c9 | |||
| 5f33abba93 | |||
| a75aef8ded | |||
| f1b78988e4 |
@@ -5,3 +5,16 @@ Shared classification, TTL and eviction rules: [caching](../common/caching.md).
|
|||||||
| Type | Mutable (auto-detected) | Immutable (auto-detected) |
|
| Type | Mutable (auto-detected) | Immutable (auto-detected) |
|
||||||
|---|---|---|
|
|---|---|---|
|
||||||
| `github_alpine` | `APKINDEX.tar.gz` (synthesized) | `.apk` (redirected) |
|
| `github_alpine` | `APKINDEX.tar.gz` (synthesized) | `.apk` (redirected) |
|
||||||
|
|
||||||
|
## Release sync
|
||||||
|
|
||||||
|
A background syncer re-scans the GitHub releases list per remote, sharing one
|
||||||
|
lease per remote across replicas through `github_alpine_sync_state`.
|
||||||
|
|
||||||
|
- **`sync_interval`** (seconds) re-checks once per interval at a stable
|
||||||
|
per-remote offset, so remotes on the same interval spread out instead of
|
||||||
|
syncing together; its first scheduled sync waits for that slot (a cold-cache
|
||||||
|
request still primes at once). Unset, it re-checks no more often than its
|
||||||
|
`mutable_ttl`.
|
||||||
|
- A failed scan keeps serving the last good index and retries with exponential
|
||||||
|
backoff.
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ Every proxied request on a remote is classified as **denied**, **immutable** or
|
|||||||
| `mutable_patterns` | Regexes forcing a path mutable. |
|
| `mutable_patterns` | Regexes forcing a path mutable. |
|
||||||
| `immutable_ttl` | Seconds an immutable path stays fresh. `0` = forever. |
|
| `immutable_ttl` | Seconds an immutable path stays fresh. `0` = forever. |
|
||||||
| `mutable_ttl` | Seconds a mutable path stays fresh. `0` also means never expires. Omitted on create, it defaults to `3600`; omitted on update, the stored value is kept. |
|
| `mutable_ttl` | Seconds a mutable path stays fresh. `0` also means never expires. Omitted on create, it defaults to `3600`; omitted on update, the stored value is kept. |
|
||||||
|
| `sync_interval` | github_rpm/github_deb/github_alpine only. Seconds between background release re-scans, at a stable per-remote offset. Unset (`0`) re-scans no more often than `mutable_ttl`. |
|
||||||
| `check_mutable` | When a mutable path's `mutable_ttl` expires, sends a conditional `HEAD` (`If-None-Match`) with the stored upstream ETag. A `304` re-arms the TTL and serves the stored copy; anything else refetches in full. The ETag is kept for `max(10 × mutable_ttl, 7 days)`, re-armed on each `304`, so idle paths age out. It is only stored when `check_mutable` is set and `mutable_ttl` is non-zero, and never for immutable paths. |
|
| `check_mutable` | When a mutable path's `mutable_ttl` expires, sends a conditional `HEAD` (`If-None-Match`) with the stored upstream ETag. A `304` re-arms the TTL and serves the stored copy; anything else refetches in full. The ETag is kept for `max(10 × mutable_ttl, 7 days)`, re-armed on each `304`, so idle paths age out. It is only stored when `check_mutable` is set and `mutable_ttl` is non-zero, and never for immutable paths. |
|
||||||
| `stale_on_error` | On an upstream network error, serve the stored copy instead of failing. |
|
| `stale_on_error` | On an upstream network error, serve the stored copy instead of failing. |
|
||||||
| `serve_stale` | rpm/deb/alpine virtual members only. When a virtual re-merges, use this member's previous index at once and refresh it in the background; the next merge picks up the refresh. The first merge still waits, and so does any merge once the previous index is over an hour old, so a member that keeps failing surfaces its error. Default `false`. |
|
| `serve_stale` | rpm/deb/alpine virtual members only. When a virtual re-merges, use this member's previous index at once and refresh it in the background; the next merge picks up the refresh. The first merge still waits, and so does any merge once the previous index is over an hour old, so a member that keeps failing surfaces its error. Default `false`. |
|
||||||
|
|||||||
@@ -5,3 +5,16 @@ Shared classification, TTL and eviction rules: [caching](../common/caching.md).
|
|||||||
| Type | Mutable (auto-detected) | Immutable (auto-detected) |
|
| Type | Mutable (auto-detected) | Immutable (auto-detected) |
|
||||||
|---|---|---|
|
|---|---|---|
|
||||||
| `github_deb` | `Packages*`, `Release*` (synthesized) | `.deb` (redirected) |
|
| `github_deb` | `Packages*`, `Release*` (synthesized) | `.deb` (redirected) |
|
||||||
|
|
||||||
|
## Release sync
|
||||||
|
|
||||||
|
A background syncer re-scans the GitHub releases list per remote, sharing one
|
||||||
|
lease per remote across replicas through `github_deb_sync_state`.
|
||||||
|
|
||||||
|
- **`sync_interval`** (seconds) re-checks once per interval at a stable
|
||||||
|
per-remote offset, so remotes on the same interval spread out instead of
|
||||||
|
syncing together; its first scheduled sync waits for that slot (a cold-cache
|
||||||
|
request still primes at once). Unset, it re-checks no more often than its
|
||||||
|
`mutable_ttl`.
|
||||||
|
- A failed scan keeps serving the last good index and retries with exponential
|
||||||
|
backoff.
|
||||||
|
|||||||
@@ -37,9 +37,9 @@ derived metadata current off the client request path:
|
|||||||
stable per-remote offset, so remotes on the same interval spread out instead
|
stable per-remote offset, so remotes on the same interval spread out instead
|
||||||
of syncing together; its first scheduled sync waits for that slot (a
|
of syncing together; its first scheduled sync waits for that slot (a
|
||||||
cold-cache request still primes at once). Unset, it re-checks no more often
|
cold-cache request still primes at once). Unset, it re-checks no more often
|
||||||
than its `mutable_ttl`. New/changed
|
than its `mutable_ttl`. New/changed assets are derived incrementally; assets
|
||||||
assets are derived incrementally; assets already cached are never re-fetched,
|
already cached are never re-fetched, and assets that disappear upstream are
|
||||||
and assets that disappear upstream are pruned.
|
pruned.
|
||||||
- **ETag / 304 conditional requests.** The releases-list `ETag` is stored per
|
- **ETag / 304 conditional requests.** The releases-list `ETag` is stored per
|
||||||
remote and sent as `If-None-Match`; a `304 Not Modified` means nothing changed
|
remote and sent as `If-None-Match`; a `304 Not Modified` means nothing changed
|
||||||
and the syncer derives nothing. GitHub does not count `304` conditional
|
and the syncer derives nothing. GitHub does not count `304` conditional
|
||||||
@@ -54,8 +54,8 @@ derived metadata current off the client request path:
|
|||||||
periodic scan a replica must atomically claim a per-remote lease
|
periodic scan a replica must atomically claim a per-remote lease
|
||||||
(`github_rpm_sync_state`: `last_synced_at`, `etag`, `sync_lease_owner`,
|
(`github_rpm_sync_state`: `last_synced_at`, `etag`, `sync_lease_owner`,
|
||||||
`sync_lease_expires`); only the winner scans. This bounds total GitHub load to
|
`sync_lease_expires`); only the winner scans. This bounds total GitHub load to
|
||||||
~once per `sync_interval` (else `mutable_ttl`) regardless of replica count, and the shared `etag`
|
~once per `sync_interval` (else `mutable_ttl`) regardless of replica count,
|
||||||
lets any replica issue the conditional request.
|
and the shared `etag` lets any replica issue the conditional request.
|
||||||
|
|
||||||
```hcl
|
```hcl
|
||||||
# Backend that serves the actual .rpm bytes from github.com.
|
# Backend that serves the actual .rpm bytes from github.com.
|
||||||
@@ -72,6 +72,7 @@ resource "artifactapi_remote_github_rpm" "acme-tools" {
|
|||||||
base_url = "https://api.github.com/repos/acme/tools" # the releases API root
|
base_url = "https://api.github.com/repos/acme/tools" # the releases API root
|
||||||
releases_remote = "github" # backend for downloads
|
releases_remote = "github" # backend for downloads
|
||||||
mutable_ttl = 3600 # release re-scan interval
|
mutable_ttl = 3600 # release re-scan interval
|
||||||
|
cache_repodata = true # keep generated repodata in memory
|
||||||
|
|
||||||
# Optional: restrict which release assets become packages (regex on filename).
|
# Optional: restrict which release assets become packages (regex on filename).
|
||||||
patterns = [".*\\.x86_64\\.rpm$", ".*\\.noarch\\.rpm$"]
|
patterns = [".*\\.x86_64\\.rpm$", ".*\\.noarch\\.rpm$"]
|
||||||
@@ -84,3 +85,13 @@ resource "artifactapi_remote_github_rpm" "acme-tools" {
|
|||||||
`dnf` config: `baseurl=https://artifactapi.example/api/v1/remote/acme-tools`.
|
`dnf` config: `baseurl=https://artifactapi.example/api/v1/remote/acme-tools`.
|
||||||
The repo is multi-arch (no `$basearch` needed) — `dnf` selects matching packages
|
The repo is multi-arch (no `$basearch` needed) — `dnf` selects matching packages
|
||||||
from the synthesized metadata.
|
from the synthesized metadata.
|
||||||
|
|
||||||
|
### Repodata cache
|
||||||
|
|
||||||
|
Repodata is regenerated from the metadata rows on every request by default. Set
|
||||||
|
`cache_repodata = true` on the remote to keep the generated `repomd.xml` and its
|
||||||
|
`primary`/`filelists`/`other` files in memory per replica. Each request checks a
|
||||||
|
cheap row fingerprint (count + newest id) in the database and rebuilds only when
|
||||||
|
it moved, so every replica picks up a scan's changes on its next request. The
|
||||||
|
fingerprint relies on only one scan writing a remote's rows at a time; in the
|
||||||
|
deployed server, the sync lease serialises scans across replicas.
|
||||||
|
|||||||
@@ -0,0 +1,57 @@
|
|||||||
|
package v2
|
||||||
|
|
||||||
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"net/http/httptest"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"git.unkin.net/unkin/artifactapi/internal/database"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestCacheRepodataRoundTrip(t *testing.T) {
|
||||||
|
if testDSN == "" {
|
||||||
|
t.Skip("Docker unavailable")
|
||||||
|
}
|
||||||
|
db, err := database.New(testDSN)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
h := NewRemotesHandler(db, nil, nil).Routes()
|
||||||
|
get := func() bool {
|
||||||
|
t.Helper()
|
||||||
|
w := httptest.NewRecorder()
|
||||||
|
h.ServeHTTP(w, httptest.NewRequest("GET", "/cache-rd", nil))
|
||||||
|
var got struct {
|
||||||
|
CacheRepodata *bool `json:"cache_repodata"`
|
||||||
|
}
|
||||||
|
if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil || got.CacheRepodata == nil {
|
||||||
|
t.Fatalf("get = %d %s", w.Code, w.Body.String())
|
||||||
|
}
|
||||||
|
return *got.CacheRepodata
|
||||||
|
}
|
||||||
|
const body = `"package_type":"github_rpm","base_url":"https://api.github.com"`
|
||||||
|
|
||||||
|
steps := []struct {
|
||||||
|
method, path, json string
|
||||||
|
want bool
|
||||||
|
}{
|
||||||
|
{"POST", "/", `{"name":"cache-rd",` + body + `}`, false},
|
||||||
|
{"PUT", "/cache-rd", `{` + body + `,"cache_repodata":true}`, true},
|
||||||
|
{"PUT", "/cache-rd", `{` + body + `,"cache_repodata":false}`, false},
|
||||||
|
}
|
||||||
|
for i, s := range steps {
|
||||||
|
if c := do(t, h, s.method, s.path, s.json); c/100 != 2 {
|
||||||
|
t.Fatalf("step %d %s %s = %d", i, s.method, s.path, c)
|
||||||
|
}
|
||||||
|
if got := get(); got != s.want {
|
||||||
|
t.Errorf("step %d: cache_repodata = %v, want %v", i, got, s.want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if c := do(t, h, "POST", "/", `{"name":"cache-rd-on",`+body+`,"cache_repodata":true}`); c != 201 {
|
||||||
|
t.Fatalf("create with cache_repodata = %d", c)
|
||||||
|
}
|
||||||
|
if r, err := db.GetRemote(t.Context(), "cache-rd-on"); err != nil || !r.CacheRepodata {
|
||||||
|
t.Fatalf("create did not store cache_repodata: %v %v", r, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -57,7 +57,7 @@ func TestMutableTTLOmittedVsZero(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// sync_interval round-trips through the API and DB on github_* remotes and is
|
// sync_interval round-trips through the API and DB on github_* remotes and is
|
||||||
// rejected when negative or on a type without a background syncer.
|
// rejected when negative, over the maximum, or on a type without a background syncer.
|
||||||
func TestSyncIntervalAPI(t *testing.T) {
|
func TestSyncIntervalAPI(t *testing.T) {
|
||||||
if testDSN == "" {
|
if testDSN == "" {
|
||||||
t.Skip("Docker unavailable")
|
t.Skip("Docker unavailable")
|
||||||
@@ -93,12 +93,20 @@ func TestSyncIntervalAPI(t *testing.T) {
|
|||||||
|
|
||||||
for _, body := range []string{
|
for _, body := range []string{
|
||||||
`{"name":"si-neg",` + gh + `,"sync_interval":-1}`,
|
`{"name":"si-neg",` + gh + `,"sync_interval":-1}`,
|
||||||
|
`{"name":"si-max",` + gh + `,"sync_interval":2592001}`,
|
||||||
|
`{"name":"si-int32",` + gh + `,"sync_interval":2147483648}`,
|
||||||
`{"name":"si-rpm","package_type":"rpm","base_url":"https://example.com/repo","sync_interval":60}`,
|
`{"name":"si-rpm","package_type":"rpm","base_url":"https://example.com/repo","sync_interval":60}`,
|
||||||
} {
|
} {
|
||||||
if c := do(t, h, "POST", "/", body); c != 400 {
|
if c := do(t, h, "POST", "/", body); c != 400 {
|
||||||
t.Errorf("%s = %d, want 400", body, c)
|
t.Errorf("%s = %d, want 400", body, c)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if c := do(t, h, "PUT", "/si-gh", `{`+gh+`,"sync_interval":-1}`); c != 400 {
|
||||||
|
t.Errorf("update negative = %d, want 400", c)
|
||||||
|
}
|
||||||
|
if got := interval("si-gh"); got != 3600 {
|
||||||
|
t.Errorf("rejected update changed sync_interval to %d", got)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestQuarantineDaysDefault(t *testing.T) {
|
func TestQuarantineDaysDefault(t *testing.T) {
|
||||||
|
|||||||
@@ -79,13 +79,19 @@ func TestRemotesCRUD(t *testing.T) {
|
|||||||
if err != nil || got.BaseURL != "https://example.com" {
|
if err != nil || got.BaseURL != "https://example.com" {
|
||||||
t.Fatalf("get: %v %v", got, err)
|
t.Fatalf("get: %v %v", got, err)
|
||||||
}
|
}
|
||||||
|
if got.CacheRepodata {
|
||||||
|
t.Error("cache_repodata should default to false")
|
||||||
|
}
|
||||||
got.BaseURL = "https://updated.example.com"
|
got.BaseURL = "https://updated.example.com"
|
||||||
|
got.CacheRepodata = true
|
||||||
|
got.SyncInterval = 600
|
||||||
|
got.ServeStale = true
|
||||||
if err := testDB.UpdateRemote(ctx(), got); err != nil {
|
if err := testDB.UpdateRemote(ctx(), got); err != nil {
|
||||||
t.Fatalf("update: %v", err)
|
t.Fatalf("update: %v", err)
|
||||||
}
|
}
|
||||||
got, _ = testDB.GetRemote(ctx(), "r-crud")
|
got, _ = testDB.GetRemote(ctx(), "r-crud")
|
||||||
if got.BaseURL != "https://updated.example.com" {
|
if got.BaseURL != "https://updated.example.com" || !got.CacheRepodata || got.SyncInterval != 600 || !got.ServeStale {
|
||||||
t.Errorf("update not applied: %v", got.BaseURL)
|
t.Errorf("update not applied: %v %v %v %v", got.BaseURL, got.CacheRepodata, got.SyncInterval, got.ServeStale)
|
||||||
}
|
}
|
||||||
list, err := testDB.ListRemotes(ctx())
|
list, err := testDB.ListRemotes(ctx())
|
||||||
if err != nil || len(list) == 0 {
|
if err != nil || len(list) == 0 {
|
||||||
@@ -487,3 +493,44 @@ func TestRPMMetadata(t *testing.T) {
|
|||||||
t.Fatalf("list rpm rows: %v %v", len(rows), err)
|
t.Fatalf("list rpm rows: %v %v", len(rows), err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestRPMMetadataFingerprint(t *testing.T) {
|
||||||
|
requireDB(t)
|
||||||
|
seedRemote(t, "r-rpm-fp")
|
||||||
|
fp := func() [2]int64 {
|
||||||
|
t.Helper()
|
||||||
|
c, m, err := testDB.RPMMetadataFingerprint(ctx(), "r-rpm-fp")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
return [2]int64{c, m}
|
||||||
|
}
|
||||||
|
insert := func(path string) {
|
||||||
|
t.Helper()
|
||||||
|
if err := testDB.InsertRPMMetadata(ctx(), &provider.RPMMetadata{
|
||||||
|
RepoName: "r-rpm-fp", FilePath: path, ContentHash: "sha256:" + path,
|
||||||
|
Name: path, Version: "1", Release: "1", Arch: "noarch",
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if got := fp(); got != [2]int64{0, 0} {
|
||||||
|
t.Fatalf("empty fingerprint = %v", got)
|
||||||
|
}
|
||||||
|
insert("a.rpm")
|
||||||
|
one := fp()
|
||||||
|
insert("a.rpm")
|
||||||
|
if fp() != one {
|
||||||
|
t.Fatal("conflicting insert changed fingerprint")
|
||||||
|
}
|
||||||
|
insert("b.rpm")
|
||||||
|
two := fp()
|
||||||
|
if err := testDB.DeleteRPMMetadata(ctx(), "r-rpm-fp", "a.rpm"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
afterDelete := fp()
|
||||||
|
if one[0] != 1 || two[0] != 2 || afterDelete[0] != 1 || afterDelete == one || afterDelete == two {
|
||||||
|
t.Fatalf("fingerprints did not track changes: %v %v %v", one, two, afterDelete)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -50,7 +50,8 @@ func (db *DB) ReleaseGitHubSyncLease(ctx context.Context, remoteName, owner stri
|
|||||||
// next_retry_at, or, with no retry pending, the remote was never synced or was
|
// next_retry_at, or, with no retry pending, the remote was never synced or was
|
||||||
// last synced before the current slot (interval set) or longer than freshness
|
// last synced before the current slot (interval set) or longer than freshness
|
||||||
// ago (otherwise). With interval set, slots start at phase past each multiple
|
// ago (otherwise). With interval set, slots start at phase past each multiple
|
||||||
// of interval and a remote created inside the current slot waits for the next.
|
// of interval and a remote created inside the current slot with no state row
|
||||||
|
// waits for the next.
|
||||||
// Slots are computed from the DB clock so replicas agree whatever their own
|
// Slots are computed from the DB clock so replicas agree whatever their own
|
||||||
// clocks say. A zero freshness and interval (prime scans) ignores the recency
|
// clocks say. A zero freshness and interval (prime scans) ignores the recency
|
||||||
// gate but still honours a pending retry's backoff. The returned etag is the
|
// gate but still honours a pending retry's backoff. The returned etag is the
|
||||||
@@ -63,7 +64,8 @@ func (db *DB) claimSyncLease(ctx context.Context, table, remoteName, owner strin
|
|||||||
SELECT r.name, $2, now() + make_interval(secs => $4)
|
SELECT r.name, $2, now() + make_interval(secs => $4)
|
||||||
FROM remotes r
|
FROM remotes r
|
||||||
WHERE r.name = $1
|
WHERE r.name = $1
|
||||||
AND ($5::float8 <= 0 OR COALESCE(r.created_at, '-infinity') < `+slot+`)
|
AND ($5::float8 <= 0 OR COALESCE(r.created_at, '-infinity') < `+slot+`
|
||||||
|
OR EXISTS (SELECT 1 FROM `+table+` e WHERE e.remote_name = r.name))
|
||||||
ON CONFLICT (remote_name) DO UPDATE
|
ON CONFLICT (remote_name) DO UPDATE
|
||||||
SET sync_lease_owner = $2,
|
SET sync_lease_owner = $2,
|
||||||
sync_lease_expires = now() + make_interval(secs => $4)
|
sync_lease_expires = now() + make_interval(secs => $4)
|
||||||
|
|||||||
@@ -400,3 +400,90 @@ func TestSyncLeaseSlot(t *testing.T) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// slotPhase returns the phase that puts the current interval slot's start
|
||||||
|
// startAgo before the DB clock.
|
||||||
|
func slotPhase(t *testing.T, interval, startAgo time.Duration) time.Duration {
|
||||||
|
t.Helper()
|
||||||
|
var epoch float64
|
||||||
|
if err := testDB.Pool.QueryRow(ctx(), `SELECT extract(epoch FROM now())::float8`).Scan(&epoch); err != nil {
|
||||||
|
t.Fatalf("db now: %v", err)
|
||||||
|
}
|
||||||
|
return time.Duration(math.Mod(epoch-startAgo.Seconds(), interval.Seconds()) * float64(time.Second))
|
||||||
|
}
|
||||||
|
|
||||||
|
func setSyncedAgo(t *testing.T, tc syncTable, name string, ago time.Duration) {
|
||||||
|
t.Helper()
|
||||||
|
if _, err := testDB.Pool.Exec(ctx(), `UPDATE `+tc.table+` SET last_synced_at = now() - make_interval(secs => $2), sync_lease_owner = '', sync_lease_expires = NULL WHERE remote_name = $1`, name, ago.Seconds()); err != nil {
|
||||||
|
t.Fatalf("set last_synced_at: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Toggling sync_interval on a remote with sync state neither re-syncs at once
|
||||||
|
// nor leaves it waiting longer than one interval (or freshness) since its last
|
||||||
|
// sync. Elapsed time is simulated by moving last_synced_at and the slot start
|
||||||
|
// back together, relative to the DB clock.
|
||||||
|
func TestSyncLeaseIntervalChange(t *testing.T) {
|
||||||
|
requireDB(t)
|
||||||
|
const (
|
||||||
|
interval = time.Hour
|
||||||
|
freshness = time.Hour
|
||||||
|
day = 24 * time.Hour
|
||||||
|
)
|
||||||
|
for _, tc := range syncTables() {
|
||||||
|
t.Run(tc.table, func(t *testing.T) {
|
||||||
|
t.Run("0 to N", func(t *testing.T) {
|
||||||
|
name := seedSynced(t, tc, day, 25*time.Minute)
|
||||||
|
if mustClaim(t, tc, name, "r1", freshness, 0, 0) {
|
||||||
|
t.Fatal("claimed inside freshness before the change")
|
||||||
|
}
|
||||||
|
if mustClaim(t, tc, name, "r1", freshness, interval, slotPhase(t, interval, 30*time.Minute)) {
|
||||||
|
t.Fatal("synced inside the current slot, yet claimed right after the change")
|
||||||
|
}
|
||||||
|
// 30m later the next slot opens; the last sync is 55m old.
|
||||||
|
setSyncedAgo(t, tc, name, 55*time.Minute)
|
||||||
|
if !mustClaim(t, tc, name, "r1", freshness, interval, slotPhase(t, interval, time.Second)) {
|
||||||
|
t.Fatal("not due at the first slot after the last sync")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
t.Run("N to 0", func(t *testing.T) {
|
||||||
|
name := seedSynced(t, tc, day, 10*time.Minute)
|
||||||
|
if mustClaim(t, tc, name, "r1", freshness, interval, slotPhase(t, interval, 30*time.Minute)) {
|
||||||
|
t.Fatal("claimed inside the slot before the change")
|
||||||
|
}
|
||||||
|
if mustClaim(t, tc, name, "r1", freshness, 0, 0) {
|
||||||
|
t.Fatal("synced inside freshness, yet claimed right after the change")
|
||||||
|
}
|
||||||
|
setSyncedAgo(t, tc, name, freshness+time.Second)
|
||||||
|
if !mustClaim(t, tc, name, "r1", freshness, 0, 0) {
|
||||||
|
t.Fatal("not due once freshness elapsed after the change")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The created_at gate only holds back the first insert of a remote created
|
||||||
|
// inside the current slot; once a state row exists the row's own sync and
|
||||||
|
// retry state decide.
|
||||||
|
func TestSyncLeaseCreatedGateInsertOnly(t *testing.T) {
|
||||||
|
requireDB(t)
|
||||||
|
const interval = time.Hour
|
||||||
|
phase := slotPhase(t, interval, 30*time.Minute)
|
||||||
|
for _, tc := range syncTables() {
|
||||||
|
t.Run(tc.table, func(t *testing.T) {
|
||||||
|
fresh := seedSynced(t, tc, 10*time.Second, neverSynced)
|
||||||
|
if mustClaim(t, tc, fresh, "r1", 0, interval, phase) {
|
||||||
|
t.Fatal("remote created inside the slot claimed with no state row")
|
||||||
|
}
|
||||||
|
|
||||||
|
name := seedSynced(t, tc, 10*time.Second, neverSynced)
|
||||||
|
if _, err := testDB.Pool.Exec(ctx(), `INSERT INTO `+tc.table+` (remote_name, sync_failures, next_retry_at) VALUES ($1, 1, now() - interval '1 second')`, name); err != nil {
|
||||||
|
t.Fatalf("seed failed prime: %v", err)
|
||||||
|
}
|
||||||
|
if !mustClaim(t, tc, name, "r1", 0, interval, phase) {
|
||||||
|
t.Fatal("created_at gate blocked a remote with a state row whose retry is due")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -15,7 +15,8 @@ const remoteCols = `name, package_type, repo_type, base_url, mirrorlist, mirror_
|
|||||||
quarantine_enabled, quarantine_days, stale_on_error,
|
quarantine_enabled, quarantine_days, stale_on_error,
|
||||||
releases_remote, managed_by,
|
releases_remote, managed_by,
|
||||||
upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout,
|
upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout,
|
||||||
serve_stale, created_at, updated_at`
|
serve_stale, cache_repodata,
|
||||||
|
created_at, updated_at`
|
||||||
|
|
||||||
// normalizeMirrorStrategy maps an empty strategy to the round_robin default so
|
// normalizeMirrorStrategy maps an empty strategy to the round_robin default so
|
||||||
// the NOT NULL mirror_strategy column always stores a canonical value.
|
// the NOT NULL mirror_strategy column always stores a canonical value.
|
||||||
@@ -35,7 +36,8 @@ func scanRemote(scanner interface{ Scan(...any) error }, r *models.Remote) error
|
|||||||
&r.QuarantineEnabled, &r.QuarantineDays, &r.StaleOnError,
|
&r.QuarantineEnabled, &r.QuarantineDays, &r.StaleOnError,
|
||||||
&r.ReleasesRemote, &r.ManagedBy,
|
&r.ReleasesRemote, &r.ManagedBy,
|
||||||
&r.UpstreamDialTimeout, &r.UpstreamTLSTimeout, &r.UpstreamResponseHeaderTimeout,
|
&r.UpstreamDialTimeout, &r.UpstreamTLSTimeout, &r.UpstreamResponseHeaderTimeout,
|
||||||
&r.ServeStale, &r.CreatedAt, &r.UpdatedAt,
|
&r.ServeStale, &r.CacheRepodata,
|
||||||
|
&r.CreatedAt, &r.UpdatedAt,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -76,8 +78,8 @@ func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error {
|
|||||||
quarantine_enabled, quarantine_days, stale_on_error,
|
quarantine_enabled, quarantine_days, stale_on_error,
|
||||||
releases_remote, managed_by,
|
releases_remote, managed_by,
|
||||||
upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout,
|
upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout,
|
||||||
mirror_strategy, sync_interval, serve_stale
|
mirror_strategy, sync_interval, serve_stale, cache_repodata
|
||||||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28)
|
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29)
|
||||||
`,
|
`,
|
||||||
r.Name, r.PackageType, r.RepoType, r.BaseURL, r.Mirrorlist, r.Description, r.Username, r.Password,
|
r.Name, r.PackageType, r.RepoType, r.BaseURL, r.Mirrorlist, r.Description, r.Username, r.Password,
|
||||||
r.ImmutableTTL, r.MutableTTL, r.CheckMutable,
|
r.ImmutableTTL, r.MutableTTL, r.CheckMutable,
|
||||||
@@ -86,7 +88,7 @@ func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error {
|
|||||||
r.QuarantineEnabled, r.QuarantineDays, r.StaleOnError,
|
r.QuarantineEnabled, r.QuarantineDays, r.StaleOnError,
|
||||||
r.ReleasesRemote, r.ManagedBy,
|
r.ReleasesRemote, r.ManagedBy,
|
||||||
r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout,
|
r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout,
|
||||||
normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval, r.ServeStale,
|
normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval, r.ServeStale, r.CacheRepodata,
|
||||||
)
|
)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -101,7 +103,7 @@ func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error {
|
|||||||
quarantine_enabled=$17, quarantine_days=$18, stale_on_error=$19,
|
quarantine_enabled=$17, quarantine_days=$18, stale_on_error=$19,
|
||||||
releases_remote=$20, managed_by=$21,
|
releases_remote=$20, managed_by=$21,
|
||||||
upstream_dial_timeout=$22, upstream_tls_timeout=$23, upstream_response_header_timeout=$24,
|
upstream_dial_timeout=$22, upstream_tls_timeout=$23, upstream_response_header_timeout=$24,
|
||||||
mirror_strategy=$26, sync_interval=$27, serve_stale=$28,
|
mirror_strategy=$26, sync_interval=$27, serve_stale=$28, cache_repodata=$29,
|
||||||
updated_at=NOW()
|
updated_at=NOW()
|
||||||
WHERE name=$1
|
WHERE name=$1
|
||||||
`,
|
`,
|
||||||
@@ -113,7 +115,7 @@ func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error {
|
|||||||
r.ReleasesRemote, r.ManagedBy,
|
r.ReleasesRemote, r.ManagedBy,
|
||||||
r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout,
|
r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout,
|
||||||
r.Mirrorlist,
|
r.Mirrorlist,
|
||||||
normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval, r.ServeStale,
|
normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval, r.ServeStale, r.CacheRepodata,
|
||||||
)
|
)
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -40,6 +40,14 @@ func (db *DB) DeleteRPMMetadata(ctx context.Context, repoName, filePath string)
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// RPMMetadataFingerprint is a cheap change detector for a repo's rows: they are
|
||||||
|
// only ever inserted or deleted, so (count, max id) moves whenever they change.
|
||||||
|
// For github_rpm remotes it relies on the sync lease serialising scans.
|
||||||
|
func (db *DB) RPMMetadataFingerprint(ctx context.Context, repoName string) (count, maxID int64, err error) {
|
||||||
|
err = db.Pool.QueryRow(ctx, `SELECT count(*), coalesce(max(id), 0) FROM rpm_metadata WHERE repo_name = $1`, repoName).Scan(&count, &maxID)
|
||||||
|
return count, maxID, err
|
||||||
|
}
|
||||||
|
|
||||||
type RPMMetadataRow struct {
|
type RPMMetadataRow struct {
|
||||||
RepoName string
|
RepoName string
|
||||||
FilePath string
|
FilePath string
|
||||||
|
|||||||
@@ -15,9 +15,11 @@ import (
|
|||||||
"regexp"
|
"regexp"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
rpmlib "github.com/cavaliergopher/rpm"
|
rpmlib "github.com/cavaliergopher/rpm"
|
||||||
|
"golang.org/x/sync/singleflight"
|
||||||
"golang.org/x/time/rate"
|
"golang.org/x/time/rate"
|
||||||
|
|
||||||
"git.unkin.net/unkin/artifactapi/internal/githubauth"
|
"git.unkin.net/unkin/artifactapi/internal/githubauth"
|
||||||
@@ -52,6 +54,9 @@ const (
|
|||||||
// retryable 503. Kept short so a client never hangs on a rate-limited derive
|
// retryable 503. Kept short so a client never hangs on a rate-limited derive
|
||||||
// of a large repo; small repos usually prime within this window.
|
// of a large repo; small repos usually prime within this window.
|
||||||
defaultColdWait = 8 * time.Second
|
defaultColdWait = 8 * time.Second
|
||||||
|
// repodataIdleTTL evicts cached indexes of remotes deleted or switched off.
|
||||||
|
repodataIdleTTL = time.Hour
|
||||||
|
repodataSweepEvery = time.Minute
|
||||||
)
|
)
|
||||||
|
|
||||||
// GitHubProvider is a metadata-only remote: it scans a GitHub repo's releases
|
// GitHubProvider is a metadata-only remote: it scans a GitHub repo's releases
|
||||||
@@ -84,6 +89,26 @@ type GitHubProvider struct {
|
|||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
scanning map[string]bool
|
scanning map[string]bool
|
||||||
lastScan map[string]time.Time
|
lastScan map[string]time.Time
|
||||||
|
|
||||||
|
// repodata caches generated indexes for cache_repodata remotes, keyed by remote name.
|
||||||
|
repodataMu sync.Mutex
|
||||||
|
repodataCache map[string]cachedRepodata
|
||||||
|
repodataGroup singleflight.Group
|
||||||
|
repodataSeq atomic.Int64
|
||||||
|
repodataSwept time.Time
|
||||||
|
}
|
||||||
|
|
||||||
|
// rpmFingerprinter is a cheap change detector over a remote's metadata rows.
|
||||||
|
// Rows are insert-or-delete only, so (count, max id) changes whenever they do.
|
||||||
|
type rpmFingerprinter interface {
|
||||||
|
RPMMetadataFingerprint(ctx context.Context, repoName string) (count, maxID int64, err error)
|
||||||
|
}
|
||||||
|
|
||||||
|
type cachedRepodata struct {
|
||||||
|
count, maxID int64
|
||||||
|
seq int64 // build order, so a slow older build never replaces a newer one
|
||||||
|
used time.Time
|
||||||
|
*repodata
|
||||||
}
|
}
|
||||||
|
|
||||||
func newGitHubProvider() *GitHubProvider {
|
func newGitHubProvider() *GitHubProvider {
|
||||||
@@ -97,6 +122,7 @@ func newGitHubProvider() *GitHubProvider {
|
|||||||
coldWait: defaultColdWait,
|
coldWait: defaultColdWait,
|
||||||
scanning: map[string]bool{},
|
scanning: map[string]bool{},
|
||||||
lastScan: map[string]time.Time{},
|
lastScan: map[string]time.Time{},
|
||||||
|
repodataCache: map[string]cachedRepodata{},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -170,6 +196,11 @@ func (p *GitHubProvider) ServeRemote(w http.ResponseWriter, r *http.Request, rem
|
|||||||
}
|
}
|
||||||
|
|
||||||
tail := strings.TrimPrefix(path, "repodata/")
|
tail := strings.TrimPrefix(path, "repodata/")
|
||||||
|
if fp, ok := store.(rpmFingerprinter); ok && remote.CacheRepodata {
|
||||||
|
p.serveCachedRepodata(w, sr, store, fp, remote.Name, tail)
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
p.dropRepodata(remote.Name)
|
||||||
lp := &Provider{}
|
lp := &Provider{}
|
||||||
switch {
|
switch {
|
||||||
case tail == "repomd.xml":
|
case tail == "repomd.xml":
|
||||||
@@ -239,7 +270,104 @@ func (p *GitHubProvider) ensurePrimed(ctx context.Context, remote models.Remote,
|
|||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// serveCachedRepodata serves a file from the generated index, rebuilding it
|
||||||
|
// only when the rows' fingerprint moved. The fingerprint is read from the shared
|
||||||
|
// database on every request, so each replica rebuilds as soon as any replica's
|
||||||
|
// scan changes rows and repomd never outlives the files it references.
|
||||||
|
func (p *GitHubProvider) serveCachedRepodata(w http.ResponseWriter, r *http.Request, store provider.RPMMetadataReader, fp rpmFingerprinter, name, tail string) {
|
||||||
|
var pick func(*repodata) []byte
|
||||||
|
contentType := "application/gzip"
|
||||||
|
switch {
|
||||||
|
case tail == "repomd.xml":
|
||||||
|
pick, contentType = func(rd *repodata) []byte { return rd.repomd }, "application/xml"
|
||||||
|
case strings.HasSuffix(tail, "-primary.xml.gz"):
|
||||||
|
pick = func(rd *repodata) []byte { return rd.primary }
|
||||||
|
case strings.HasSuffix(tail, "-filelists.xml.gz"):
|
||||||
|
pick = func(rd *repodata) []byte { return rd.filelists }
|
||||||
|
case strings.HasSuffix(tail, "-other.xml.gz"):
|
||||||
|
pick = func(rd *repodata) []byte { return rd.other }
|
||||||
|
default:
|
||||||
|
http.Error(w, "not found", http.StatusNotFound)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
rd, err := p.cachedRepodata(r.Context(), store, fp, name)
|
||||||
|
if err != nil {
|
||||||
|
writeMetadataError(w, name, err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
w.Header().Set("Content-Type", contentType)
|
||||||
|
w.WriteHeader(http.StatusOK)
|
||||||
|
_, _ = w.Write(pick(rd))
|
||||||
|
}
|
||||||
|
|
||||||
|
// cachedRepodata returns the index for the current fingerprint, building it at
|
||||||
|
// most once per fingerprint. ctx is ServeRemote's detached serve context, so a
|
||||||
|
// leader's client disconnect cannot fail the build for coalesced waiters.
|
||||||
|
func (p *GitHubProvider) cachedRepodata(ctx context.Context, store provider.RPMMetadataReader, fp rpmFingerprinter, name string) (*repodata, error) {
|
||||||
|
seq := p.repodataSeq.Add(1)
|
||||||
|
count, maxID, err := fp.RPMMetadataFingerprint(ctx, name)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
now := time.Now()
|
||||||
|
p.repodataMu.Lock()
|
||||||
|
p.sweepRepodataLocked(now)
|
||||||
|
c, ok := p.repodataCache[name]
|
||||||
|
if ok {
|
||||||
|
c.used = now
|
||||||
|
p.repodataCache[name] = c
|
||||||
|
}
|
||||||
|
p.repodataMu.Unlock()
|
||||||
|
if ok && c.count == count && c.maxID == maxID {
|
||||||
|
return c.repodata, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
v, err, _ := p.repodataGroup.Do(fmt.Sprintf("%s\x00%d\x00%d", name, count, maxID), func() (any, error) {
|
||||||
|
metas, err := store.ListRPMMetadataEntries(ctx, name)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
// Rows read after the fingerprint can only be newer than it, so the
|
||||||
|
// worst case is one extra rebuild on the next request.
|
||||||
|
rd := buildRepodata(metas)
|
||||||
|
p.repodataMu.Lock()
|
||||||
|
if cur, ok := p.repodataCache[name]; !ok || cur.seq < seq {
|
||||||
|
p.repodataCache[name] = cachedRepodata{count: count, maxID: maxID, seq: seq, used: time.Now(), repodata: rd}
|
||||||
|
}
|
||||||
|
p.repodataMu.Unlock()
|
||||||
|
return rd, nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return v.(*repodata), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (p *GitHubProvider) dropRepodata(name string) {
|
||||||
|
p.repodataMu.Lock()
|
||||||
|
delete(p.repodataCache, name)
|
||||||
|
p.sweepRepodataLocked(time.Now())
|
||||||
|
p.repodataMu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
// sweepRepodataLocked evicts idle entries, at most once per repodataSweepEvery.
|
||||||
|
func (p *GitHubProvider) sweepRepodataLocked(now time.Time) {
|
||||||
|
if now.Sub(p.repodataSwept) < repodataSweepEvery {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
p.repodataSwept = now
|
||||||
|
for name, c := range p.repodataCache {
|
||||||
|
if now.Sub(c.used) > repodataIdleTTL {
|
||||||
|
delete(p.repodataCache, name)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (p *GitHubProvider) cacheEmpty(ctx context.Context, store provider.RemoteMetadataStore, name string) bool {
|
func (p *GitHubProvider) cacheEmpty(ctx context.Context, store provider.RemoteMetadataStore, name string) bool {
|
||||||
|
if fp, ok := store.(rpmFingerprinter); ok {
|
||||||
|
count, _, err := fp.RPMMetadataFingerprint(ctx, name)
|
||||||
|
return err == nil && count == 0
|
||||||
|
}
|
||||||
rows, err := store.ListRPMMetadataEntries(ctx, name)
|
rows, err := store.ListRPMMetadataEntries(ctx, name)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// Treat a failed read as "not empty" so a transient DB error becomes a
|
// Treat a failed read as "not empty" so a transient DB error becomes a
|
||||||
|
|||||||
@@ -0,0 +1,365 @@
|
|||||||
|
package rpm
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"errors"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"runtime"
|
||||||
|
"strings"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"git.unkin.net/unkin/artifactapi/internal/provider"
|
||||||
|
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||||
|
)
|
||||||
|
|
||||||
|
// fpStore adds a fingerprint and a full-read counter to fakeStore.
|
||||||
|
type fpStore struct {
|
||||||
|
*fakeStore
|
||||||
|
lists, fps int
|
||||||
|
seq int64
|
||||||
|
fpErr, listErr error
|
||||||
|
gate chan struct{} // when set, full reads block until it is closed
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *fpStore) ListRPMMetadataEntries(ctx context.Context, name string) ([]provider.RPMMetadata, error) {
|
||||||
|
s.mu.Lock()
|
||||||
|
s.lists++
|
||||||
|
gate, err := s.gate, s.listErr
|
||||||
|
s.mu.Unlock()
|
||||||
|
if gate != nil {
|
||||||
|
<-gate
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return s.fakeStore.ListRPMMetadataEntries(ctx, name)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *fpStore) InsertRPMMetadata(ctx context.Context, m *provider.RPMMetadata) error {
|
||||||
|
s.mu.Lock()
|
||||||
|
_, exists := s.rows[m.FilePath]
|
||||||
|
s.mu.Unlock()
|
||||||
|
if err := s.fakeStore.InsertRPMMetadata(ctx, m); err != nil || exists {
|
||||||
|
return err // ON CONFLICT DO NOTHING leaves max(id) unchanged
|
||||||
|
}
|
||||||
|
s.mu.Lock()
|
||||||
|
s.seq++
|
||||||
|
s.mu.Unlock()
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *fpStore) RPMMetadataFingerprint(_ context.Context, _ string) (int64, int64, error) {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
s.fps++
|
||||||
|
return int64(len(s.rows)), s.seq, s.fpErr
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *fpStore) fpCount() int {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
return s.fps
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *fpStore) listCount() int {
|
||||||
|
s.mu.Lock()
|
||||||
|
defer s.mu.Unlock()
|
||||||
|
return s.lists
|
||||||
|
}
|
||||||
|
|
||||||
|
func newCacheFixture(t *testing.T, cache bool) (*GitHubProvider, *fpStore, models.Remote) {
|
||||||
|
t.Helper()
|
||||||
|
remote := models.Remote{Name: "cached-rpm", PackageType: models.PackageGitHubRPM, MutableTTL: 3600, CacheRepodata: cache}
|
||||||
|
p := newGitHubProvider()
|
||||||
|
p.lastScan[remote.Name] = time.Now()
|
||||||
|
store := &fpStore{fakeStore: newFakeStore()}
|
||||||
|
addPkg(t, store, "alpha")
|
||||||
|
return p, store, remote
|
||||||
|
}
|
||||||
|
|
||||||
|
func addPkg(t *testing.T, store *fpStore, name string) {
|
||||||
|
t.Helper()
|
||||||
|
if err := store.InsertRPMMetadata(context.Background(), &provider.RPMMetadata{
|
||||||
|
RepoName: "cached-rpm", FilePath: name + "-1-1.noarch.rpm", ContentHash: "sha256:" + name,
|
||||||
|
Name: name, Version: "1", Release: "1", Arch: "noarch", Group: "Unspecified",
|
||||||
|
Files: []provider.RPMFile{{Path: "/usr/bin/" + name}},
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func serveReq(t *testing.T, p *GitHubProvider, remote models.Remote, store *fpStore, req *http.Request, path string) *httptest.ResponseRecorder {
|
||||||
|
t.Helper()
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
if !p.ServeRemote(rec, req, remote, path, "https://proxy.example", store) {
|
||||||
|
t.Errorf("%s not handled", path)
|
||||||
|
}
|
||||||
|
return rec
|
||||||
|
}
|
||||||
|
|
||||||
|
func serve(t *testing.T, p *GitHubProvider, remote models.Remote, store *fpStore, path string) []byte {
|
||||||
|
t.Helper()
|
||||||
|
rec := serveReq(t, p, remote, store, httptest.NewRequest(http.MethodGet, "/x", nil), path)
|
||||||
|
if rec.Code != http.StatusOK {
|
||||||
|
t.Fatalf("%s: code %d: %s", path, rec.Code, rec.Body.String())
|
||||||
|
}
|
||||||
|
return rec.Body.Bytes()
|
||||||
|
}
|
||||||
|
|
||||||
|
var repodataPaths = []string{
|
||||||
|
"repodata/repomd.xml",
|
||||||
|
"repodata/x-primary.xml.gz",
|
||||||
|
"repodata/x-filelists.xml.gz",
|
||||||
|
"repodata/x-other.xml.gz",
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCachedRepodataHitSkipsRegeneration(t *testing.T) {
|
||||||
|
p, store, remote := newCacheFixture(t, true)
|
||||||
|
|
||||||
|
repomd := serve(t, p, remote, store, "repodata/repomd.xml")
|
||||||
|
primary := serve(t, p, remote, store, "repodata/x-primary.xml.gz")
|
||||||
|
if n := store.listCount(); n != 1 {
|
||||||
|
t.Fatalf("full metadata reads = %d, want 1", n)
|
||||||
|
}
|
||||||
|
if !strings.Contains(string(repomd), sha256Hex(primary)+"-primary.xml.gz") {
|
||||||
|
t.Fatalf("cached repomd does not reference cached primary:\n%s", repomd)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCachedRepodataRebuildsOnDataChange(t *testing.T) {
|
||||||
|
p, store, remote := newCacheFixture(t, true)
|
||||||
|
before := serve(t, p, remote, store, "repodata/repomd.xml")
|
||||||
|
|
||||||
|
addPkg(t, store, "beta")
|
||||||
|
after := serve(t, p, remote, store, "repodata/repomd.xml")
|
||||||
|
primary := serve(t, p, remote, store, "repodata/x-primary.xml.gz")
|
||||||
|
|
||||||
|
if string(before) == string(after) {
|
||||||
|
t.Fatal("repomd unchanged after metadata change")
|
||||||
|
}
|
||||||
|
if !strings.Contains(string(after), sha256Hex(primary)+"-primary.xml.gz") {
|
||||||
|
t.Fatal("rebuilt repomd does not reference rebuilt primary")
|
||||||
|
}
|
||||||
|
if !strings.Contains(gunzip(t, primary), "<name>beta</name>") {
|
||||||
|
t.Fatal("rebuilt primary missing new package")
|
||||||
|
}
|
||||||
|
if n := store.listCount(); n != 2 {
|
||||||
|
t.Fatalf("full metadata reads = %d, want 2", n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestUncachedRepodataMatchesCachedAndReadsEveryTime(t *testing.T) {
|
||||||
|
p, store, remote := newCacheFixture(t, true)
|
||||||
|
cached := map[string]string{}
|
||||||
|
for _, path := range repodataPaths {
|
||||||
|
cached[path] = string(serve(t, p, remote, store, path))
|
||||||
|
}
|
||||||
|
|
||||||
|
remote.CacheRepodata = false
|
||||||
|
reads := store.listCount()
|
||||||
|
for _, path := range repodataPaths {
|
||||||
|
if got := string(serve(t, p, remote, store, path)); got != cached[path] {
|
||||||
|
t.Fatalf("%s differs between cached and uncached", path)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if n := store.listCount() - reads; n != len(repodataPaths) {
|
||||||
|
t.Fatalf("uncached full reads = %d, want %d", n, len(repodataPaths))
|
||||||
|
}
|
||||||
|
if _, ok := p.repodataCache[remote.Name]; ok {
|
||||||
|
t.Fatal("disabling cache_repodata should drop the cached index")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCachedRepodataRebuildsOnSameCountReinsert(t *testing.T) {
|
||||||
|
p, store, remote := newCacheFixture(t, true)
|
||||||
|
addPkg(t, store, "alpha") // conflicting insert: fingerprint must not move
|
||||||
|
serve(t, p, remote, store, "repodata/repomd.xml")
|
||||||
|
serve(t, p, remote, store, "repodata/repomd.xml")
|
||||||
|
if n := store.listCount(); n != 1 {
|
||||||
|
t.Fatalf("full reads after conflicting insert = %d, want 1", n)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err := store.DeleteRPMMetadata(context.Background(), remote.Name, "alpha-1-1.noarch.rpm"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
addPkg(t, store, "alpha")
|
||||||
|
serve(t, p, remote, store, "repodata/repomd.xml")
|
||||||
|
if n := store.listCount(); n != 2 {
|
||||||
|
t.Fatalf("full reads after same-count reinsert = %d, want 2", n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCachedRepodataEvictsIdleRemotes(t *testing.T) {
|
||||||
|
p, store, remote := newCacheFixture(t, true)
|
||||||
|
serve(t, p, remote, store, "repodata/repomd.xml")
|
||||||
|
p.repodataCache["gone"] = cachedRepodata{used: time.Now().Add(-2 * repodataIdleTTL), repodata: &repodata{}}
|
||||||
|
p.repodataSwept = time.Time{}
|
||||||
|
|
||||||
|
serve(t, p, remote, store, "repodata/repomd.xml")
|
||||||
|
if _, ok := p.repodataCache["gone"]; ok {
|
||||||
|
t.Fatal("idle remote's index not evicted")
|
||||||
|
}
|
||||||
|
if _, ok := p.repodataCache[remote.Name]; !ok {
|
||||||
|
t.Fatal("active remote's index evicted")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func waitFor(t *testing.T, cond func() bool) {
|
||||||
|
t.Helper()
|
||||||
|
deadline := time.Now().Add(10 * time.Second)
|
||||||
|
for !cond() {
|
||||||
|
if time.Now().After(deadline) {
|
||||||
|
t.Fatal("condition not reached")
|
||||||
|
}
|
||||||
|
time.Sleep(time.Millisecond)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// sfWaiters counts goroutines parked on an in-flight singleflight call.
|
||||||
|
func sfWaiters() int {
|
||||||
|
buf := make([]byte, 1<<20)
|
||||||
|
for {
|
||||||
|
n := runtime.Stack(buf, true)
|
||||||
|
if n < len(buf) {
|
||||||
|
buf = buf[:n]
|
||||||
|
break
|
||||||
|
}
|
||||||
|
buf = make([]byte, 2*len(buf))
|
||||||
|
}
|
||||||
|
waiters := 0
|
||||||
|
for _, g := range strings.Split(string(buf), "\n\n") {
|
||||||
|
if strings.Contains(g, "sync.(*WaitGroup).Wait") && strings.Contains(g, "singleflight.(*Group).Do(") {
|
||||||
|
waiters++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return waiters
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCachedRepodataCoalescesConcurrentBuilds(t *testing.T) {
|
||||||
|
p, store, remote := newCacheFixture(t, true)
|
||||||
|
gate := make(chan struct{})
|
||||||
|
store.gate = gate
|
||||||
|
const n = 8
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
codes := make([]int, n)
|
||||||
|
for i := range n {
|
||||||
|
wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
defer wg.Done()
|
||||||
|
codes[i] = serveReq(t, p, remote, store, httptest.NewRequest(http.MethodGet, "/x", nil), repodataPaths[i%len(repodataPaths)]).Code
|
||||||
|
}()
|
||||||
|
}
|
||||||
|
waitFor(t, func() bool { return store.listCount() == 1 && sfWaiters() == n-1 })
|
||||||
|
close(gate)
|
||||||
|
wg.Wait()
|
||||||
|
for i, c := range codes {
|
||||||
|
if c != http.StatusOK {
|
||||||
|
t.Errorf("request %d: code %d", i, c)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if got := store.listCount(); got != 1 {
|
||||||
|
t.Fatalf("full metadata reads = %d, want 1", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCachedRepodataLeaderCancelDoesNotFailWaiters(t *testing.T) {
|
||||||
|
p, store, remote := newCacheFixture(t, true)
|
||||||
|
gate := make(chan struct{})
|
||||||
|
store.gate = gate
|
||||||
|
leaderCtx, cancelLeader := context.WithCancel(context.Background())
|
||||||
|
results := make(chan int, 2)
|
||||||
|
go func() {
|
||||||
|
req := httptest.NewRequest(http.MethodGet, "/x", nil).WithContext(leaderCtx)
|
||||||
|
results <- serveReq(t, p, remote, store, req, "repodata/repomd.xml").Code
|
||||||
|
}()
|
||||||
|
waitFor(t, func() bool { return store.listCount() == 1 })
|
||||||
|
go func() {
|
||||||
|
results <- serveReq(t, p, remote, store, httptest.NewRequest(http.MethodGet, "/x", nil), "repodata/repomd.xml").Code
|
||||||
|
}()
|
||||||
|
waitFor(t, func() bool { return sfWaiters() == 1 })
|
||||||
|
cancelLeader()
|
||||||
|
close(gate)
|
||||||
|
for range 2 {
|
||||||
|
if c := <-results; c != http.StatusOK {
|
||||||
|
t.Fatalf("code %d after leader cancel, want 200", c)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if got := store.listCount(); got != 1 {
|
||||||
|
t.Fatalf("full metadata reads = %d, want 1", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCachedRepodataOlderBuildDoesNotOverwriteNewer(t *testing.T) {
|
||||||
|
p, store, remote := newCacheFixture(t, true)
|
||||||
|
newer := &repodata{}
|
||||||
|
p.repodataCache[remote.Name] = cachedRepodata{count: 99, maxID: 99, seq: 1 << 40, used: time.Now(), repodata: newer}
|
||||||
|
serve(t, p, remote, store, "repodata/repomd.xml")
|
||||||
|
if p.repodataCache[remote.Name].repodata != newer {
|
||||||
|
t.Fatal("older build replaced a newer cache entry")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCachedRepodataErrors(t *testing.T) {
|
||||||
|
secret := errors.New("pq: password authentication failed for user secret")
|
||||||
|
for _, tc := range []struct {
|
||||||
|
name string
|
||||||
|
fpErr, listErr error
|
||||||
|
code int
|
||||||
|
}{
|
||||||
|
{"fingerprint", secret, nil, http.StatusInternalServerError},
|
||||||
|
{"list", nil, secret, http.StatusInternalServerError},
|
||||||
|
{"fingerprint deadline", context.DeadlineExceeded, nil, http.StatusServiceUnavailable},
|
||||||
|
{"list canceled", nil, context.Canceled, http.StatusServiceUnavailable},
|
||||||
|
} {
|
||||||
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
p, store, remote := newCacheFixture(t, true)
|
||||||
|
store.fpErr, store.listErr = tc.fpErr, tc.listErr
|
||||||
|
rec := serveReq(t, p, remote, store, httptest.NewRequest(http.MethodGet, "/x", nil), "repodata/repomd.xml")
|
||||||
|
if rec.Code != tc.code {
|
||||||
|
t.Fatalf("code %d, want %d", rec.Code, tc.code)
|
||||||
|
}
|
||||||
|
if strings.Contains(rec.Body.String(), "secret") {
|
||||||
|
t.Fatalf("DB error leaked: %q", rec.Body.String())
|
||||||
|
}
|
||||||
|
if retry := rec.Header().Get("Retry-After"); (tc.code == http.StatusServiceUnavailable) != (retry != "") {
|
||||||
|
t.Fatalf("Retry-After = %q for %d", retry, tc.code)
|
||||||
|
}
|
||||||
|
if _, ok := p.repodataCache[remote.Name]; ok {
|
||||||
|
t.Fatal("failed build was cached")
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCachedRepodataUnknownTail404sWithoutDB(t *testing.T) {
|
||||||
|
p, store, remote := newCacheFixture(t, true)
|
||||||
|
rec := serveReq(t, p, remote, store, httptest.NewRequest(http.MethodGet, "/x", nil), "repodata/bogus.txt")
|
||||||
|
if rec.Code != http.StatusNotFound {
|
||||||
|
t.Fatalf("code %d, want 404", rec.Code)
|
||||||
|
}
|
||||||
|
if store.fpCount() != 0 || store.listCount() != 0 {
|
||||||
|
t.Fatalf("unknown tail hit the DB: fps=%d lists=%d", store.fpCount(), store.listCount())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestCacheEmptyUsesFingerprint(t *testing.T) {
|
||||||
|
p := newGitHubProvider()
|
||||||
|
store := &fpStore{fakeStore: newFakeStore()}
|
||||||
|
if !p.cacheEmpty(context.Background(), store, "cached-rpm") {
|
||||||
|
t.Fatal("empty store not reported empty")
|
||||||
|
}
|
||||||
|
addPkg(t, store, "alpha")
|
||||||
|
if p.cacheEmpty(context.Background(), store, "cached-rpm") {
|
||||||
|
t.Fatal("populated store reported empty")
|
||||||
|
}
|
||||||
|
store.fpErr = errors.New("db down")
|
||||||
|
if p.cacheEmpty(context.Background(), store, "cached-rpm") {
|
||||||
|
t.Fatal("fingerprint error must not report empty")
|
||||||
|
}
|
||||||
|
if store.listCount() != 0 {
|
||||||
|
t.Fatalf("cacheEmpty did full reads: %d", store.listCount())
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -250,36 +250,52 @@ func (p *Provider) GenerateLocalIndex(ctx context.Context, files provider.FileSt
|
|||||||
func readMetadataEntries(w http.ResponseWriter, r *http.Request, reader provider.RPMMetadataReader, repoName string) ([]provider.RPMMetadata, bool) {
|
func readMetadataEntries(w http.ResponseWriter, r *http.Request, reader provider.RPMMetadataReader, repoName string) ([]provider.RPMMetadata, bool) {
|
||||||
metas, err := reader.ListRPMMetadataEntries(r.Context(), repoName)
|
metas, err := reader.ListRPMMetadataEntries(r.Context(), repoName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
writeMetadataError(w, repoName, err)
|
||||||
slog.Warn("rpm: metadata read canceled", "repo", repoName, "error", err)
|
|
||||||
http.Error(w, "metadata read canceled", http.StatusServiceUnavailable)
|
|
||||||
return nil, false
|
|
||||||
}
|
|
||||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
|
||||||
return nil, false
|
return nil, false
|
||||||
}
|
}
|
||||||
return metas, true
|
return metas, true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// writeMetadataError maps a metadata read failure to a response without leaking
|
||||||
|
// DB error text: a canceled/expired context is a retryable 503, anything else a 500.
|
||||||
|
func writeMetadataError(w http.ResponseWriter, repoName string, err error) {
|
||||||
|
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
||||||
|
slog.Warn("rpm: metadata read canceled", "repo", repoName, "error", err)
|
||||||
|
w.Header().Set("Retry-After", "5")
|
||||||
|
http.Error(w, "metadata read canceled", http.StatusServiceUnavailable)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
slog.Error("rpm: metadata read failed", "repo", repoName, "error", err)
|
||||||
|
http.Error(w, "metadata read failed", http.StatusInternalServerError)
|
||||||
|
}
|
||||||
|
|
||||||
func (p *Provider) serveRepomd(w http.ResponseWriter, r *http.Request, reader provider.RPMMetadataReader, repoName string) {
|
func (p *Provider) serveRepomd(w http.ResponseWriter, r *http.Request, reader provider.RPMMetadataReader, repoName string) {
|
||||||
metas, ok := readMetadataEntries(w, r, reader, repoName)
|
metas, ok := readMetadataEntries(w, r, reader, repoName)
|
||||||
if !ok {
|
if !ok {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
primary := generatePrimaryXMLGZ(metas)
|
|
||||||
filelists := generateFilelistsXMLGZ(metas)
|
|
||||||
other := generateOtherXMLGZ(metas)
|
|
||||||
|
|
||||||
primaryHash := sha256Hex(primary)
|
|
||||||
filelistsHash := sha256Hex(filelists)
|
|
||||||
otherHash := sha256Hex(other)
|
|
||||||
|
|
||||||
repomd := generateRepomd(repomdRevision(metas), primaryHash, len(primary), filelistsHash, len(filelists), otherHash, len(other))
|
|
||||||
|
|
||||||
w.Header().Set("Content-Type", "application/xml")
|
w.Header().Set("Content-Type", "application/xml")
|
||||||
w.WriteHeader(http.StatusOK)
|
w.WriteHeader(http.StatusOK)
|
||||||
w.Write(repomd)
|
_, _ = w.Write(buildRepodata(metas).repomd)
|
||||||
|
}
|
||||||
|
|
||||||
|
// repodata is a complete generated index: repomd.xml plus the three gz files it references.
|
||||||
|
type repodata struct {
|
||||||
|
repomd, primary, filelists, other []byte
|
||||||
|
}
|
||||||
|
|
||||||
|
func buildRepodata(metas []provider.RPMMetadata) *repodata {
|
||||||
|
rd := &repodata{
|
||||||
|
primary: generatePrimaryXMLGZ(metas),
|
||||||
|
filelists: generateFilelistsXMLGZ(metas),
|
||||||
|
other: generateOtherXMLGZ(metas),
|
||||||
|
}
|
||||||
|
rd.repomd = generateRepomd(repomdRevision(metas),
|
||||||
|
sha256Hex(rd.primary), len(rd.primary),
|
||||||
|
sha256Hex(rd.filelists), len(rd.filelists),
|
||||||
|
sha256Hex(rd.other), len(rd.other))
|
||||||
|
return rd
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *Provider) servePrimary(w http.ResponseWriter, r *http.Request, reader provider.RPMMetadataReader, repoName string) {
|
func (p *Provider) servePrimary(w http.ResponseWriter, r *http.Request, reader provider.RPMMetadataReader, repoName string) {
|
||||||
|
|||||||
@@ -510,3 +510,9 @@ func TestSyncerScheduledPrime(t *testing.T) {
|
|||||||
t.Fatalf("cold-cache prime: attempts=%d freshness=%v interval=%v, want 1/0/0", store.attempts, store.freshness, store.interval)
|
t.Fatalf("cold-cache prime: attempts=%d freshness=%v interval=%v, want 1/0/0", store.attempts, store.freshness, store.interval)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestSyncLeaseOutlivesScanTimeout(t *testing.T) {
|
||||||
|
if syncLeaseDuration <= defaultScanTimeout {
|
||||||
|
t.Fatalf("syncLeaseDuration %v must exceed defaultScanTimeout %v so scans stay serialised", syncLeaseDuration, defaultScanTimeout)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -59,6 +59,7 @@ func (m *model) viewRemoteDetail() string {
|
|||||||
sb.WriteString(fmt.Sprintf(" Revalidation: %v\n", r.CheckMutable))
|
sb.WriteString(fmt.Sprintf(" Revalidation: %v\n", r.CheckMutable))
|
||||||
sb.WriteString(fmt.Sprintf(" Stale on Error: %v\n", r.StaleOnError))
|
sb.WriteString(fmt.Sprintf(" Stale on Error: %v\n", r.StaleOnError))
|
||||||
fmt.Fprintf(&sb, " Serve Stale: %v\n", r.ServeStale)
|
fmt.Fprintf(&sb, " Serve Stale: %v\n", r.ServeStale)
|
||||||
|
fmt.Fprintf(&sb, " Cache Repodata: %v\n", r.CacheRepodata)
|
||||||
|
|
||||||
if len(r.Patterns) > 0 {
|
if len(r.Patterns) > 0 {
|
||||||
sb.WriteString(fmt.Sprintf(" Patterns: %s\n", strings.Join(r.Patterns, ", ")))
|
sb.WriteString(fmt.Sprintf(" Patterns: %s\n", strings.Join(r.Patterns, ", ")))
|
||||||
@@ -108,8 +109,12 @@ func (m *model) viewVirtualsList() string {
|
|||||||
}
|
}
|
||||||
|
|
||||||
for i, v := range m.virtuals {
|
for i, v := range m.virtuals {
|
||||||
line := fmt.Sprintf(" %-25s %-12s %d members %s",
|
parallel := "-"
|
||||||
v.Name, v.PackageType, len(v.Members), v.Description)
|
if v.ParallelFetch {
|
||||||
|
parallel = "parallel"
|
||||||
|
}
|
||||||
|
line := fmt.Sprintf(" %-25s %-12s %3d members %-8s %s",
|
||||||
|
v.Name, v.PackageType, len(v.Members), parallel, v.Description)
|
||||||
if i == m.cursor {
|
if i == m.cursor {
|
||||||
sb.WriteString(selStyle.Render(line))
|
sb.WriteString(selStyle.Render(line))
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
@@ -548,8 +548,8 @@ func (e *Engine) fetchAlpineMember(ctx context.Context, name, dir string) (*Alpi
|
|||||||
// the background, one refresh per member at a time; the next merge picks it up.
|
// the background, one refresh per member at a time; the next merge picks it up.
|
||||||
func staleMember[T any](ctx context.Context, e *Engine, key, name string, fetch func(context.Context) (*T, error)) (*T, error) {
|
func staleMember[T any](ctx context.Context, e *Engine, key, name string, fetch func(context.Context) (*T, error)) (*T, error) {
|
||||||
sk := key + "\x00" + name
|
sk := key + "\x00" + name
|
||||||
remote, err := e.getRemote(ctx, name)
|
remote, lookupErr := e.getRemote(ctx, name)
|
||||||
if err != nil || !remote.ServeStale {
|
if lookupErr == nil && !remote.ServeStale {
|
||||||
e.mu.Lock()
|
e.mu.Lock()
|
||||||
delete(e.stale, sk)
|
delete(e.stale, sk)
|
||||||
e.mu.Unlock()
|
e.mu.Unlock()
|
||||||
@@ -560,6 +560,11 @@ func staleMember[T any](ctx context.Context, e *Engine, key, name string, fetch
|
|||||||
e.stale = map[string]*staleEntry{}
|
e.stale = map[string]*staleEntry{}
|
||||||
}
|
}
|
||||||
prev := e.stale[sk]
|
prev := e.stale[sk]
|
||||||
|
// A failed lookup serves an existing copy but never creates one.
|
||||||
|
if lookupErr != nil && (prev == nil || prev.m == nil) {
|
||||||
|
e.mu.Unlock()
|
||||||
|
return fetch(ctx)
|
||||||
|
}
|
||||||
// A cold fetch holds a placeholder so eviction or toggle-off during it is seen.
|
// A cold fetch holds a placeholder so eviction or toggle-off during it is seen.
|
||||||
token := prev
|
token := prev
|
||||||
if prev == nil || prev.m == nil {
|
if prev == nil || prev.m == nil {
|
||||||
|
|||||||
@@ -55,9 +55,14 @@ func fetchRepomd(t *testing.T, e *Engine) string {
|
|||||||
|
|
||||||
func TestServeStaleMemberDoesNotBlockMerge(t *testing.T) {
|
func TestServeStaleMemberDoesNotBlockMerge(t *testing.T) {
|
||||||
release := make(chan struct{})
|
release := make(chan struct{})
|
||||||
|
refreshing := make(chan struct{}, 1)
|
||||||
var slow atomic.Bool
|
var slow atomic.Bool
|
||||||
e, calls := staleEngine(true, func() *RPMMember {
|
e, calls := staleEngine(true, func() *RPMMember {
|
||||||
if slow.Load() {
|
if slow.Load() {
|
||||||
|
select {
|
||||||
|
case refreshing <- struct{}{}:
|
||||||
|
default:
|
||||||
|
}
|
||||||
<-release
|
<-release
|
||||||
return rpmPkgMember("a", "2")
|
return rpmPkgMember("a", "2")
|
||||||
}
|
}
|
||||||
@@ -78,12 +83,17 @@ func TestServeStaleMemberDoesNotBlockMerge(t *testing.T) {
|
|||||||
t.Fatal("merge blocked on a serve_stale member refresh")
|
t.Fatal("merge blocked on a serve_stale member refresh")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
select {
|
||||||
|
case <-refreshing:
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
t.Fatal("background refresh never started")
|
||||||
|
}
|
||||||
if n := calls.Load(); n != 2 {
|
if n := calls.Load(); n != 2 {
|
||||||
t.Fatalf("member fetched %d times, want 2 (cold + one single-flighted refresh)", n)
|
t.Fatalf("member fetched %d times, want 2 (cold + one single-flighted refresh)", n)
|
||||||
}
|
}
|
||||||
|
|
||||||
close(release)
|
close(release)
|
||||||
deadline := time.Now().Add(time.Second)
|
deadline := time.Now().Add(5 * time.Second)
|
||||||
for fetchRepomd(t, e) == v1 {
|
for fetchRepomd(t, e) == v1 {
|
||||||
if time.Now().After(deadline) {
|
if time.Now().After(deadline) {
|
||||||
t.Fatal("background refresh never reached a merge")
|
t.Fatal("background refresh never reached a merge")
|
||||||
@@ -328,3 +338,128 @@ func TestMergePrunesExpiredCopiesOfOtherMembers(t *testing.T) {
|
|||||||
t.Fatal("any merge should prune expired copies of removed members")
|
t.Fatal("any merge should prune expired copies of removed members")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func primary(m any) string { return string(m.(*RPMMember).Data["primary"]) }
|
||||||
|
|
||||||
|
func TestServeStaleLookupErrorKeepsCopy(t *testing.T) {
|
||||||
|
release := make(chan struct{})
|
||||||
|
started := make(chan struct{}, 1)
|
||||||
|
var slow atomic.Bool
|
||||||
|
e, calls := staleEngine(true, func() *RPMMember {
|
||||||
|
if slow.Load() {
|
||||||
|
select {
|
||||||
|
case started <- struct{}{}:
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
<-release
|
||||||
|
return rpmPkgMember("a", "2")
|
||||||
|
}
|
||||||
|
return rpmPkgMember("a", "1")
|
||||||
|
})
|
||||||
|
v1 := fetchRepomd(t, e)
|
||||||
|
slow.Store(true)
|
||||||
|
e.getRemote = func(context.Context, string) (*models.Remote, error) { return nil, errors.New("db down") }
|
||||||
|
|
||||||
|
done := make(chan string)
|
||||||
|
go func() { done <- fetchRepomd(t, e) }()
|
||||||
|
select {
|
||||||
|
case got := <-done:
|
||||||
|
if got != v1 {
|
||||||
|
t.Fatal("a failed remote lookup should serve the previous copy")
|
||||||
|
}
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("merge blocked on the member fetch instead of serving the copy")
|
||||||
|
}
|
||||||
|
select {
|
||||||
|
case <-started:
|
||||||
|
case <-time.After(time.Second):
|
||||||
|
t.Fatal("no background refresh started")
|
||||||
|
}
|
||||||
|
if n := calls.Load(); n != 2 {
|
||||||
|
t.Fatalf("member fetched %d times, want 2 (cold + one background refresh)", n)
|
||||||
|
}
|
||||||
|
close(release)
|
||||||
|
waitRefresh(e, rpmStaleKey)
|
||||||
|
e.mu.Lock()
|
||||||
|
defer e.mu.Unlock()
|
||||||
|
if se := e.stale[rpmStaleKey]; se == nil || se.m == nil || primary(se.m) != primary(rpmPkgMember("a", "2")) {
|
||||||
|
t.Fatal("the background refresh should replace the copy")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestServeStaleLookupErrorDuringColdFetch(t *testing.T) {
|
||||||
|
e := fakeEngine(nil)
|
||||||
|
e.getRemote = staleRemote
|
||||||
|
ctx := context.Background()
|
||||||
|
entered, release := make(chan struct{}), make(chan struct{})
|
||||||
|
first := make(chan *RPMMember)
|
||||||
|
go func() {
|
||||||
|
m, err := staleMember(ctx, e, "rpm\x00v", "a", func(context.Context) (*RPMMember, error) {
|
||||||
|
close(entered)
|
||||||
|
<-release
|
||||||
|
return rpmPkgMember("a", "1"), nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Error(err)
|
||||||
|
}
|
||||||
|
first <- m
|
||||||
|
}()
|
||||||
|
<-entered
|
||||||
|
e.mu.Lock()
|
||||||
|
placeholder := e.stale[rpmStaleKey]
|
||||||
|
e.mu.Unlock()
|
||||||
|
if placeholder == nil || placeholder.m != nil {
|
||||||
|
t.Fatal("a cold fetch should hold an empty placeholder")
|
||||||
|
}
|
||||||
|
|
||||||
|
e.getRemote = func(context.Context, string) (*models.Remote, error) { return nil, errors.New("db down") }
|
||||||
|
m, err := staleMember(ctx, e, "rpm\x00v", "a", func(context.Context) (*RPMMember, error) { return rpmPkgMember("a", "2"), nil })
|
||||||
|
if err != nil || primary(m) != primary(rpmPkgMember("a", "2")) {
|
||||||
|
t.Fatalf("a failed lookup during a cold fetch should fetch directly, got %v", err)
|
||||||
|
}
|
||||||
|
e.mu.Lock()
|
||||||
|
if e.stale[rpmStaleKey] != placeholder || placeholder.m != nil {
|
||||||
|
e.mu.Unlock()
|
||||||
|
t.Fatal("a failed lookup must not touch the in-flight placeholder")
|
||||||
|
}
|
||||||
|
e.mu.Unlock()
|
||||||
|
|
||||||
|
close(release)
|
||||||
|
if got := <-first; got == nil || primary(got) != primary(rpmPkgMember("a", "1")) {
|
||||||
|
t.Fatal("the cold fetch should return its own result")
|
||||||
|
}
|
||||||
|
e.mu.Lock()
|
||||||
|
defer e.mu.Unlock()
|
||||||
|
if se := e.stale[rpmStaleKey]; se == nil || se.m == nil || primary(se.m) != primary(rpmPkgMember("a", "1")) {
|
||||||
|
t.Fatal("the cold fetch should store its own result")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestServeStaleLookupErrorColdFetchStoresNothing(t *testing.T) {
|
||||||
|
e, calls := staleEngine(true, func() *RPMMember { return rpmPkgMember("a", "1") })
|
||||||
|
e.getRemote = func(context.Context, string) (*models.Remote, error) { return nil, errors.New("db down") }
|
||||||
|
fetchRepomd(t, e)
|
||||||
|
if calls.Load() != 1 {
|
||||||
|
t.Fatalf("cold member with a failed lookup should block on one fetch, got %d", calls.Load())
|
||||||
|
}
|
||||||
|
e.mu.Lock()
|
||||||
|
defer e.mu.Unlock()
|
||||||
|
if _, ok := e.stale[rpmStaleKey]; ok {
|
||||||
|
t.Fatal("a failed lookup must not create a copy")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNonServeStaleLookupErrorStoresNothing(t *testing.T) {
|
||||||
|
e, calls := staleEngine(false, func() *RPMMember { return rpmPkgMember("a", "1") })
|
||||||
|
fetchRepomd(t, e)
|
||||||
|
e.getRemote = func(context.Context, string) (*models.Remote, error) { return nil, errors.New("db down") }
|
||||||
|
fetchRepomd(t, e)
|
||||||
|
if calls.Load() != 2 {
|
||||||
|
t.Fatalf("each merge should fetch the member, got %d", calls.Load())
|
||||||
|
}
|
||||||
|
e.mu.Lock()
|
||||||
|
defer e.mu.Unlock()
|
||||||
|
if _, ok := e.stale[rpmStaleKey]; ok {
|
||||||
|
t.Fatal("a failed lookup must not create a copy for a non-serve_stale member")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -452,7 +452,7 @@ func TestMergeRPMPassesParallelFetchAndKeepsMemberOrder(t *testing.T) {
|
|||||||
if name == "a" {
|
if name == "a" {
|
||||||
time.Sleep(20 * time.Millisecond)
|
time.Sleep(20 * time.Millisecond)
|
||||||
}
|
}
|
||||||
return &RPMMember{RemoteName: name, Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", "1", name+"sum", name+".rpm"))}}, nil
|
return &RPMMember{RemoteName: name, Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo-"+name, "1", name+"sum", name+".rpm"))}}, nil
|
||||||
}
|
}
|
||||||
virt := rpmVirt("a", "b")
|
virt := rpmVirt("a", "b")
|
||||||
virt.ParallelFetch = true
|
virt.ParallelFetch = true
|
||||||
@@ -465,7 +465,7 @@ func TestMergeRPMPassesParallelFetchAndKeepsMemberOrder(t *testing.T) {
|
|||||||
}
|
}
|
||||||
primary := regexp.MustCompile(`repodata/[0-9a-f]+-primary\.xml\.gz`).FindString(string(repo.Repomd))
|
primary := regexp.MustCompile(`repodata/[0-9a-f]+-primary\.xml\.gz`).FindString(string(repo.Repomd))
|
||||||
body := string(gunzip(t, repo.Files[primary]))
|
body := string(gunzip(t, repo.Files[primary]))
|
||||||
if a, b := strings.Index(body, "a.rpm"), strings.Index(body, "b.rpm"); a < 0 || b >= 0 && b < a {
|
if a, b := strings.Index(body, "a.rpm"), strings.Index(body, "b.rpm"); a < 0 || b < a {
|
||||||
t.Fatalf("member priority order lost:\n%s", body)
|
t.Fatalf("member priority order lost:\n%s", body)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1 @@
|
|||||||
|
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS cache_repodata BOOLEAN NOT NULL DEFAULT FALSE;
|
||||||
+11
-4
@@ -84,6 +84,10 @@ type Remote struct {
|
|||||||
ReleasesRemote string `json:"releases_remote,omitempty"`
|
ReleasesRemote string `json:"releases_remote,omitempty"`
|
||||||
ManagedBy string `json:"managed_by,omitempty"`
|
ManagedBy string `json:"managed_by,omitempty"`
|
||||||
|
|
||||||
|
// CacheRepodata keeps a github_rpm remote's generated repodata in memory,
|
||||||
|
// rebuilding it only when its metadata rows change.
|
||||||
|
CacheRepodata bool `json:"cache_repodata"`
|
||||||
|
|
||||||
CreatedAt time.Time `json:"created_at"`
|
CreatedAt time.Time `json:"created_at"`
|
||||||
UpdatedAt time.Time `json:"updated_at"`
|
UpdatedAt time.Time `json:"updated_at"`
|
||||||
}
|
}
|
||||||
@@ -185,11 +189,14 @@ func syncPhase(name string, period time.Duration) time.Duration {
|
|||||||
return time.Duration(h.Sum64() % uint64(period))
|
return time.Duration(h.Sum64() % uint64(period))
|
||||||
}
|
}
|
||||||
|
|
||||||
// ValidateSyncInterval rejects a negative sync_interval, or one set on a type
|
// MaxSyncInterval caps sync_interval at 30 days (seconds).
|
||||||
// without a background syncer.
|
const MaxSyncInterval = 30 * 24 * 60 * 60
|
||||||
|
|
||||||
|
// ValidateSyncInterval rejects a sync_interval outside 0..MaxSyncInterval, or
|
||||||
|
// one set on a type without a background syncer.
|
||||||
func (r *Remote) ValidateSyncInterval() error {
|
func (r *Remote) ValidateSyncInterval() error {
|
||||||
if r.SyncInterval < 0 {
|
if r.SyncInterval < 0 || r.SyncInterval > MaxSyncInterval {
|
||||||
return fmt.Errorf("sync_interval must be >= 0, got %d", r.SyncInterval)
|
return fmt.Errorf("sync_interval must be between 0 and %d, got %d", MaxSyncInterval, r.SyncInterval)
|
||||||
}
|
}
|
||||||
switch r.PackageType {
|
switch r.PackageType {
|
||||||
case PackageGitHubRPM, PackageGitHubDeb, PackageGitHubAlpine:
|
case PackageGitHubRPM, PackageGitHubDeb, PackageGitHubAlpine:
|
||||||
|
|||||||
@@ -153,6 +153,8 @@ func TestValidateSyncInterval(t *testing.T) {
|
|||||||
{PackageGitHubDeb, 3600, true},
|
{PackageGitHubDeb, 3600, true},
|
||||||
{PackageGitHubAlpine, 3600, true},
|
{PackageGitHubAlpine, 3600, true},
|
||||||
{PackageGitHubRPM, -1, false},
|
{PackageGitHubRPM, -1, false},
|
||||||
|
{PackageGitHubRPM, MaxSyncInterval, true},
|
||||||
|
{PackageGitHubRPM, MaxSyncInterval + 1, false},
|
||||||
{PackageRPM, 0, true},
|
{PackageRPM, 0, true},
|
||||||
{PackageRPM, 60, false},
|
{PackageRPM, 60, false},
|
||||||
} {
|
} {
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ export interface Remote {
|
|||||||
quarantine_days: number;
|
quarantine_days: number;
|
||||||
stale_on_error: boolean;
|
stale_on_error: boolean;
|
||||||
serve_stale: boolean;
|
serve_stale: boolean;
|
||||||
|
cache_repodata: boolean;
|
||||||
releases_remote: string;
|
releases_remote: string;
|
||||||
managed_by: string;
|
managed_by: string;
|
||||||
created_at: string;
|
created_at: string;
|
||||||
@@ -31,6 +32,7 @@ export interface Virtual {
|
|||||||
description: string;
|
description: string;
|
||||||
members: string[];
|
members: string[];
|
||||||
managed_by: string;
|
managed_by: string;
|
||||||
|
parallel_fetch: boolean;
|
||||||
created_at: string;
|
created_at: string;
|
||||||
updated_at: string;
|
updated_at: string;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -52,6 +52,8 @@ export function RemoteDetail() {
|
|||||||
<dd>{remote.stale_on_error ? 'enabled' : 'disabled'}</dd>
|
<dd>{remote.stale_on_error ? 'enabled' : 'disabled'}</dd>
|
||||||
<dt>Serve Stale</dt>
|
<dt>Serve Stale</dt>
|
||||||
<dd>{remote.serve_stale ? 'enabled' : 'disabled'}</dd>
|
<dd>{remote.serve_stale ? 'enabled' : 'disabled'}</dd>
|
||||||
|
<dt>Cache Repodata</dt>
|
||||||
|
<dd>{remote.cache_repodata ? 'enabled' : 'disabled'}</dd>
|
||||||
{remote.releases_remote && (
|
{remote.releases_remote && (
|
||||||
<>
|
<>
|
||||||
<dt>Releases Remote</dt>
|
<dt>Releases Remote</dt>
|
||||||
|
|||||||
+23
-21
@@ -26,6 +26,8 @@ export function Virtuals() {
|
|||||||
.finally(() => setLoading(false));
|
.finally(() => setLoading(false));
|
||||||
}, []);
|
}, []);
|
||||||
|
|
||||||
|
const selected = virtuals.find(x => x.name === expanded);
|
||||||
|
|
||||||
function memberLink(name: string) {
|
function memberLink(name: string) {
|
||||||
const remote = remoteMap[name];
|
const remote = remoteMap[name];
|
||||||
if (remote?.repo_type === 'local') {
|
if (remote?.repo_type === 'local') {
|
||||||
@@ -67,6 +69,13 @@ export function Virtuals() {
|
|||||||
header: 'Description',
|
header: 'Description',
|
||||||
render: (v: Virtual) => v.description || <span className="text-muted">—</span>,
|
render: (v: Virtual) => v.description || <span className="text-muted">—</span>,
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
key: 'parallel',
|
||||||
|
header: 'Parallel',
|
||||||
|
render: (v: Virtual) =>
|
||||||
|
v.parallel_fetch ? <Badge variant="yellow">parallel</Badge> : <span className="text-muted">—</span>,
|
||||||
|
width: '100px',
|
||||||
|
},
|
||||||
{
|
{
|
||||||
key: 'managed',
|
key: 'managed',
|
||||||
header: 'Managed',
|
header: 'Managed',
|
||||||
@@ -81,30 +90,23 @@ export function Virtuals() {
|
|||||||
/>
|
/>
|
||||||
)}
|
)}
|
||||||
|
|
||||||
{expanded && (
|
{selected && (
|
||||||
<div className="virtual-detail-panel">
|
<div className="virtual-detail-panel">
|
||||||
<h3 className="section-label">Members of {expanded}</h3>
|
<h3 className="section-label">Members of {selected.name}</h3>
|
||||||
<ul className="member-list">
|
<ul className="member-list">
|
||||||
{virtuals
|
{selected.members?.map((m, i) => {
|
||||||
.find(v => v.name === expanded)
|
const remote = remoteMap[m];
|
||||||
?.members?.map((m, i) => {
|
const typeLabel = remote?.repo_type === 'local' ? 'local' : 'remote';
|
||||||
const remote = remoteMap[m];
|
return (
|
||||||
const typeLabel = remote?.repo_type === 'local' ? 'local' : 'remote';
|
<li key={m}>
|
||||||
return (
|
<span className="member-priority">{i + 1}</span>
|
||||||
<li key={m}>
|
<Link to={memberLink(m)} className="mono">{m}</Link>
|
||||||
<span className="member-priority">{i + 1}</span>
|
<Badge variant={typeLabel === 'local' ? 'yellow' : 'default'}>{typeLabel}</Badge>
|
||||||
<Link to={memberLink(m)} className="mono">{m}</Link>
|
</li>
|
||||||
<Badge variant={typeLabel === 'local' ? 'yellow' : 'default'}>{typeLabel}</Badge>
|
);
|
||||||
</li>
|
})}
|
||||||
);
|
|
||||||
})}
|
|
||||||
</ul>
|
</ul>
|
||||||
{(() => {
|
<UsageInstructions packageType={selected.package_type} repoClass="virtual" name={selected.name} defaultOpen />
|
||||||
const v = virtuals.find(x => x.name === expanded);
|
|
||||||
return v ? (
|
|
||||||
<UsageInstructions packageType={v.package_type} repoClass="virtual" name={v.name} defaultOpen />
|
|
||||||
) : null;
|
|
||||||
})()}
|
|
||||||
</div>
|
</div>
|
||||||
)}
|
)}
|
||||||
</div>
|
</div>
|
||||||
|
|||||||
Reference in New Issue
Block a user