Merge master
This commit is contained in:
@@ -15,6 +15,7 @@ Every proxied request on a remote is classified as **denied**, **immutable** or
|
||||
| `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. |
|
||||
| `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. |
|
||||
| `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`. |
|
||||
| `quarantine_enabled` / `quarantine_days` | Refuses immutable artifacts published less than `quarantine_days` (default `3`) ago. See [Quarantine](#quarantine). |
|
||||
| `mirrorlist` / `mirror_strategy` | rpm/deb/alpine only. Upstream pool is `[base_url] + mirrorlist`; `round_robin` (default) rotates the starting mirror, `least_conn` starts at the one with the fewest in-flight requests. Network errors and 5xx fail over to the next mirror; other statuses return as-is. |
|
||||
|
||||
|
||||
@@ -15,7 +15,7 @@ const remoteCols = `name, package_type, repo_type, base_url, mirrorlist, mirror_
|
||||
quarantine_enabled, quarantine_days, stale_on_error,
|
||||
releases_remote, managed_by,
|
||||
upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout,
|
||||
created_at, updated_at`
|
||||
serve_stale, created_at, updated_at`
|
||||
|
||||
// normalizeMirrorStrategy maps an empty strategy to the round_robin default so
|
||||
// the NOT NULL mirror_strategy column always stores a canonical value.
|
||||
@@ -35,7 +35,7 @@ func scanRemote(scanner interface{ Scan(...any) error }, r *models.Remote) error
|
||||
&r.QuarantineEnabled, &r.QuarantineDays, &r.StaleOnError,
|
||||
&r.ReleasesRemote, &r.ManagedBy,
|
||||
&r.UpstreamDialTimeout, &r.UpstreamTLSTimeout, &r.UpstreamResponseHeaderTimeout,
|
||||
&r.CreatedAt, &r.UpdatedAt,
|
||||
&r.ServeStale, &r.CreatedAt, &r.UpdatedAt,
|
||||
)
|
||||
}
|
||||
|
||||
@@ -76,8 +76,8 @@ func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error {
|
||||
quarantine_enabled, quarantine_days, stale_on_error,
|
||||
releases_remote, managed_by,
|
||||
upstream_dial_timeout, upstream_tls_timeout, upstream_response_header_timeout,
|
||||
mirror_strategy, sync_interval
|
||||
) 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)
|
||||
mirror_strategy, sync_interval, serve_stale
|
||||
) 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)
|
||||
`,
|
||||
r.Name, r.PackageType, r.RepoType, r.BaseURL, r.Mirrorlist, r.Description, r.Username, r.Password,
|
||||
r.ImmutableTTL, r.MutableTTL, r.CheckMutable,
|
||||
@@ -86,7 +86,7 @@ func (db *DB) CreateRemote(ctx context.Context, r *models.Remote) error {
|
||||
r.QuarantineEnabled, r.QuarantineDays, r.StaleOnError,
|
||||
r.ReleasesRemote, r.ManagedBy,
|
||||
r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout,
|
||||
normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval,
|
||||
normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval, r.ServeStale,
|
||||
)
|
||||
return err
|
||||
}
|
||||
@@ -101,7 +101,7 @@ func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error {
|
||||
quarantine_enabled=$17, quarantine_days=$18, stale_on_error=$19,
|
||||
releases_remote=$20, managed_by=$21,
|
||||
upstream_dial_timeout=$22, upstream_tls_timeout=$23, upstream_response_header_timeout=$24,
|
||||
mirror_strategy=$26, sync_interval=$27,
|
||||
mirror_strategy=$26, sync_interval=$27, serve_stale=$28,
|
||||
updated_at=NOW()
|
||||
WHERE name=$1
|
||||
`,
|
||||
@@ -113,7 +113,7 @@ func (db *DB) UpdateRemote(ctx context.Context, r *models.Remote) error {
|
||||
r.ReleasesRemote, r.ManagedBy,
|
||||
r.UpstreamDialTimeout, r.UpstreamTLSTimeout, r.UpstreamResponseHeaderTimeout,
|
||||
r.Mirrorlist,
|
||||
normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval,
|
||||
normalizeMirrorStrategy(r.MirrorStrategy), r.SyncInterval, r.ServeStale,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -416,6 +416,25 @@ func TestServerRemoteUpdateAndVirtualCRUD(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestServerRemoteServeStaleRoundTrip(t *testing.T) {
|
||||
requireStack(t)
|
||||
if resp, b := req(t, "POST", "/api/v2/remotes", `{"name":"srv-ss","package_type":"rpm","repo_type":"remote","base_url":"https://a.example.com","serve_stale":true}`); resp.StatusCode != 201 {
|
||||
t.Fatalf("create remote: %d %s", resp.StatusCode, b)
|
||||
}
|
||||
defer req(t, "DELETE", "/api/v2/remotes/srv-ss", "")
|
||||
get := func(want string) {
|
||||
t.Helper()
|
||||
if _, b := req(t, "GET", "/api/v2/remotes/srv-ss", ""); !strings.Contains(string(b), want) {
|
||||
t.Errorf("get remote: want %s in %s", want, b)
|
||||
}
|
||||
}
|
||||
get(`"serve_stale":true`)
|
||||
if resp, b := req(t, "PUT", "/api/v2/remotes/srv-ss", `{"package_type":"rpm","base_url":"https://a.example.com"}`); resp.StatusCode != 200 {
|
||||
t.Fatalf("update remote: %d %s", resp.StatusCode, b)
|
||||
}
|
||||
get(`"serve_stale":false`)
|
||||
}
|
||||
|
||||
func TestServerLocalRemoveAndMissing(t *testing.T) {
|
||||
requireStack(t)
|
||||
req(t, "POST", "/api/v2/remotes", `{"name":"srv-rm","package_type":"generic","repo_type":"local"}`)
|
||||
|
||||
@@ -58,6 +58,7 @@ func (m *model) viewRemoteDetail() string {
|
||||
sb.WriteString(fmt.Sprintf(" Mutable TTL: %ds\n", r.MutableTTL))
|
||||
sb.WriteString(fmt.Sprintf(" Revalidation: %v\n", r.CheckMutable))
|
||||
sb.WriteString(fmt.Sprintf(" Stale on Error: %v\n", r.StaleOnError))
|
||||
fmt.Fprintf(&sb, " Serve Stale: %v\n", r.ServeStale)
|
||||
|
||||
if len(r.Patterns) > 0 {
|
||||
sb.WriteString(fmt.Sprintf(" Patterns: %s\n", strings.Join(r.Patterns, ", ")))
|
||||
|
||||
@@ -57,8 +57,20 @@ type Engine struct {
|
||||
sf singleflight.Group
|
||||
mu sync.Mutex
|
||||
gens map[string]*repoGen
|
||||
// stale holds the last good fetch of each serve_stale member, keyed by
|
||||
// merge key and member name.
|
||||
stale map[string]*staleEntry
|
||||
}
|
||||
|
||||
type staleEntry struct {
|
||||
m any
|
||||
at time.Time
|
||||
}
|
||||
|
||||
// maxStaleAge bounds how old a serve_stale member copy may be before merges
|
||||
// block on the member again and surface its error.
|
||||
const maxStaleAge = time.Hour
|
||||
|
||||
// mergedRepo is a merged index document plus the files it references.
|
||||
type mergedRepo interface {
|
||||
index() []byte
|
||||
@@ -328,6 +340,7 @@ func (e *Engine) merged(ctx context.Context, key string, merge func(context.Cont
|
||||
e.mu.Unlock()
|
||||
|
||||
v, err, _ := e.sf.Do(key, func() (any, error) {
|
||||
e.pruneStale()
|
||||
repo, err := merge(context.WithoutCancel(ctx))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -356,7 +369,9 @@ func (e *Engine) merged(ctx context.Context, key string, merge func(context.Cont
|
||||
}
|
||||
|
||||
func (e *Engine) mergeRPM(ctx context.Context, virt models.Virtual) (*RPMRepo, error) {
|
||||
members, err := fetchMembers(virt, func(name string) (*RPMMember, error) { return e.rpmMember(ctx, name, virt.ParallelFetch) })
|
||||
members, err := fetchMembers(virt, func(name string) (*RPMMember, error) {
|
||||
return staleMember(ctx, e, "rpm\x00"+virt.Name, name, func(ctx context.Context) (*RPMMember, error) { return e.rpmMember(ctx, name, virt.ParallelFetch) })
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -405,7 +420,9 @@ func (e *Engine) fetchDeb(ctx context.Context, virt models.Virtual, reqPath stri
|
||||
}
|
||||
|
||||
repo, err := e.merged(ctx, key, func(ctx context.Context) (mergedRepo, error) {
|
||||
members, err := fetchMembers(virt, func(name string) (*DebMember, error) { return e.debMember(ctx, name, base, virt.ParallelFetch) })
|
||||
members, err := fetchMembers(virt, func(name string) (*DebMember, error) {
|
||||
return staleMember(ctx, e, key, name, func(ctx context.Context) (*DebMember, error) { return e.debMember(ctx, name, base, virt.ParallelFetch) })
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -452,8 +469,11 @@ func (e *Engine) fetchDebMember(ctx context.Context, name, base string, parallel
|
||||
}
|
||||
|
||||
func (e *Engine) mergedAlpine(ctx context.Context, virt models.Virtual, dir string) (mergedRepo, error) {
|
||||
repo, err := e.merged(ctx, "alpine\x00"+virt.Name+"\x00"+dir, func(ctx context.Context) (mergedRepo, error) {
|
||||
members, err := fetchMembers(virt, func(name string) (*AlpineMember, error) { return e.apkMember(ctx, name, dir) })
|
||||
key := "alpine\x00" + virt.Name + "\x00" + dir
|
||||
repo, err := e.merged(ctx, key, func(ctx context.Context) (mergedRepo, error) {
|
||||
members, err := fetchMembers(virt, func(name string) (*AlpineMember, error) {
|
||||
return staleMember(ctx, e, key, name, func(ctx context.Context) (*AlpineMember, error) { return e.apkMember(ctx, name, dir) })
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -523,6 +543,69 @@ func (e *Engine) fetchAlpineMember(ctx context.Context, name, dir string) (*Alpi
|
||||
return &AlpineMember{RemoteName: name, Index: idx}, nil
|
||||
}
|
||||
|
||||
// staleMember fetches member name for the merge under key. A serve_stale member
|
||||
// with a good fetch younger than maxStaleAge returns it at once and refreshes in
|
||||
// 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) {
|
||||
sk := key + "\x00" + name
|
||||
remote, err := e.getRemote(ctx, name)
|
||||
if err != nil || !remote.ServeStale {
|
||||
e.mu.Lock()
|
||||
delete(e.stale, sk)
|
||||
e.mu.Unlock()
|
||||
return fetch(ctx)
|
||||
}
|
||||
e.mu.Lock()
|
||||
if e.stale == nil {
|
||||
e.stale = map[string]*staleEntry{}
|
||||
}
|
||||
prev := e.stale[sk]
|
||||
// A cold fetch holds a placeholder so eviction or toggle-off during it is seen.
|
||||
token := prev
|
||||
if prev == nil || prev.m == nil {
|
||||
prev, token = nil, &staleEntry{at: time.Now()}
|
||||
e.stale[sk] = token
|
||||
}
|
||||
e.mu.Unlock()
|
||||
refresh := func() (any, error) {
|
||||
m, err := fetch(ctx)
|
||||
e.mu.Lock()
|
||||
// An evicted or replaced entry means this refresh is no longer wanted.
|
||||
if e.stale[sk] == token {
|
||||
if err == nil {
|
||||
e.stale[sk] = &staleEntry{m: m, at: time.Now()}
|
||||
} else if prev == nil {
|
||||
delete(e.stale, sk)
|
||||
}
|
||||
}
|
||||
e.mu.Unlock()
|
||||
return m, err
|
||||
}
|
||||
if prev == nil {
|
||||
m, err := refresh()
|
||||
return m.(*T), err
|
||||
}
|
||||
ch := e.sf.DoChan("stale\x00"+sk, refresh)
|
||||
go func() {
|
||||
if r := <-ch; r.Err != nil {
|
||||
slog.Warn("virtual member background refresh failed", "member", name, "error", r.Err)
|
||||
}
|
||||
}()
|
||||
return prev.m.(*T), nil
|
||||
}
|
||||
|
||||
// pruneStale drops serve_stale copies past maxStaleAge, including those of
|
||||
// removed members and virtuals.
|
||||
func (e *Engine) pruneStale() {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
for k, se := range e.stale {
|
||||
if time.Since(se.at) >= maxStaleAge {
|
||||
delete(e.stale, k)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// evictLRU drops the least recently used merges under prefix beyond keep.
|
||||
func (e *Engine) evictLRU(prefix string, keep int) {
|
||||
e.mu.Lock()
|
||||
@@ -536,6 +619,11 @@ func (e *Engine) evictLRU(prefix string, keep int) {
|
||||
slices.SortFunc(keys, func(a, b string) int { return e.gens[b].used.Compare(e.gens[a].used) })
|
||||
for _, k := range keys[min(len(keys), keep):] {
|
||||
delete(e.gens, k)
|
||||
for sk := range e.stale {
|
||||
if strings.HasPrefix(sk, k+"\x00") {
|
||||
delete(e.stale, sk)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,330 @@
|
||||
package virtual
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.unkin.net/unkin/artifactapi/pkg/models"
|
||||
)
|
||||
|
||||
func rpmPkgMember(name, ver string) *RPMMember {
|
||||
return &RPMMember{RemoteName: name, Data: map[string][]byte{"primary": primaryXML(primaryPkgXML("foo", ver, "h"+ver, "foo-"+ver+".rpm"))}}
|
||||
}
|
||||
|
||||
// staleEngine re-merges on every request; member "a" is fetched through fetch
|
||||
// and has serve_stale set to stale.
|
||||
func staleEngine(stale bool, fetch func() *RPMMember) (*Engine, *atomic.Int32) {
|
||||
e, calls, _ := staleEngineErr(stale, func() (*RPMMember, error) { return fetch(), nil })
|
||||
return e, calls
|
||||
}
|
||||
|
||||
func staleEngineErr(stale bool, fetch func() (*RPMMember, error)) (*Engine, *atomic.Int32, *atomic.Bool) {
|
||||
var calls atomic.Int32
|
||||
var flag atomic.Bool
|
||||
flag.Store(stale)
|
||||
e := fakeEngine(nil)
|
||||
e.getRemote = func(_ context.Context, name string) (*models.Remote, error) {
|
||||
return &models.Remote{Name: name, RepoType: models.RepoTypeRemote, ServeStale: flag.Load()}, nil
|
||||
}
|
||||
e.rpmMember = func(_ context.Context, name string, _ bool) (*RPMMember, error) {
|
||||
calls.Add(1)
|
||||
return fetch()
|
||||
}
|
||||
return e, &calls, &flag
|
||||
}
|
||||
|
||||
// waitRefresh blocks until the background refresh of sk, if any, has finished.
|
||||
func waitRefresh(e *Engine, sk string) {
|
||||
_, _, _ = e.sf.Do("stale\x00"+sk, func() (any, error) { return nil, nil })
|
||||
}
|
||||
|
||||
const rpmStaleKey = "rpm\x00v\x00a"
|
||||
|
||||
func fetchRepomd(t *testing.T, e *Engine) string {
|
||||
t.Helper()
|
||||
b, _, err := e.Fetch(context.Background(), rpmVirt("a"), "repodata/repomd.xml", "")
|
||||
if err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
func TestServeStaleMemberDoesNotBlockMerge(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
var slow atomic.Bool
|
||||
e, calls := staleEngine(true, func() *RPMMember {
|
||||
if slow.Load() {
|
||||
<-release
|
||||
return rpmPkgMember("a", "2")
|
||||
}
|
||||
return rpmPkgMember("a", "1")
|
||||
})
|
||||
v1 := fetchRepomd(t, e)
|
||||
slow.Store(true)
|
||||
|
||||
for range 3 {
|
||||
done := make(chan string)
|
||||
go func() { done <- fetchRepomd(t, e) }()
|
||||
select {
|
||||
case got := <-done:
|
||||
if got != v1 {
|
||||
t.Fatal("stale merge should serve the previous member copy")
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("merge blocked on a serve_stale member refresh")
|
||||
}
|
||||
}
|
||||
if n := calls.Load(); n != 2 {
|
||||
t.Fatalf("member fetched %d times, want 2 (cold + one single-flighted refresh)", n)
|
||||
}
|
||||
|
||||
close(release)
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for fetchRepomd(t, e) == v1 {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("background refresh never reached a merge")
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
|
||||
func assertMergeBlocks(t *testing.T, e *Engine, release chan struct{}) {
|
||||
t.Helper()
|
||||
done := make(chan struct{})
|
||||
go func() { fetchRepomd(t, e); close(done) }()
|
||||
select {
|
||||
case <-done:
|
||||
t.Fatal("merge returned before the member fetch finished")
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
}
|
||||
close(release)
|
||||
<-done
|
||||
}
|
||||
|
||||
func TestNonServeStaleMemberBlocksMerge(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
var slow atomic.Bool
|
||||
e, _ := staleEngine(false, func() *RPMMember {
|
||||
if slow.Load() {
|
||||
<-release
|
||||
}
|
||||
return rpmPkgMember("a", "1")
|
||||
})
|
||||
fetchRepomd(t, e)
|
||||
slow.Store(true)
|
||||
assertMergeBlocks(t, e, release)
|
||||
}
|
||||
|
||||
func TestServeStaleColdStartBlocksMerge(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
e, _ := staleEngine(true, func() *RPMMember {
|
||||
<-release
|
||||
return rpmPkgMember("a", "1")
|
||||
})
|
||||
assertMergeBlocks(t, e, release)
|
||||
}
|
||||
|
||||
func TestServeStaleRefreshErrorKeepsCopy(t *testing.T) {
|
||||
var down atomic.Bool
|
||||
e, _, _ := staleEngineErr(true, func() (*RPMMember, error) {
|
||||
if down.Load() {
|
||||
return nil, errors.New("upstream down")
|
||||
}
|
||||
return rpmPkgMember("a", "1"), nil
|
||||
})
|
||||
v1 := fetchRepomd(t, e)
|
||||
down.Store(true)
|
||||
for range 3 {
|
||||
if got := fetchRepomd(t, e); got != v1 {
|
||||
t.Fatal("a failed refresh should keep serving the previous copy")
|
||||
}
|
||||
waitRefresh(e, rpmStaleKey)
|
||||
}
|
||||
}
|
||||
|
||||
func TestServeStaleMaxAgeFallsBackToFetch(t *testing.T) {
|
||||
var down atomic.Bool
|
||||
e, _, _ := staleEngineErr(true, func() (*RPMMember, error) {
|
||||
if down.Load() {
|
||||
return nil, errors.New("upstream down")
|
||||
}
|
||||
return rpmPkgMember("a", "1"), nil
|
||||
})
|
||||
fetchRepomd(t, e)
|
||||
down.Store(true)
|
||||
e.mu.Lock()
|
||||
e.stale[rpmStaleKey].at = time.Now().Add(-maxStaleAge)
|
||||
e.mu.Unlock()
|
||||
if _, _, err := e.Fetch(context.Background(), rpmVirt("a"), "repodata/repomd.xml", ""); err == nil {
|
||||
t.Fatal("a copy past maxStaleAge should not be served")
|
||||
}
|
||||
if len(e.stale) != 0 {
|
||||
t.Fatal("an expired copy should be dropped")
|
||||
}
|
||||
}
|
||||
|
||||
func TestServeStaleToggleOffClearsCopy(t *testing.T) {
|
||||
var down atomic.Bool
|
||||
e, _, flag := staleEngineErr(true, func() (*RPMMember, error) {
|
||||
if down.Load() {
|
||||
return nil, errors.New("upstream down")
|
||||
}
|
||||
return rpmPkgMember("a", "1"), nil
|
||||
})
|
||||
fetchRepomd(t, e)
|
||||
flag.Store(false)
|
||||
fetchRepomd(t, e)
|
||||
if len(e.stale) != 0 {
|
||||
t.Fatal("turning serve_stale off should drop the copy")
|
||||
}
|
||||
flag.Store(true)
|
||||
down.Store(true)
|
||||
if _, _, err := e.Fetch(context.Background(), rpmVirt("a"), "repodata/repomd.xml", ""); err == nil {
|
||||
t.Fatal("re-enabling serve_stale must not resurrect the old copy")
|
||||
}
|
||||
}
|
||||
|
||||
func staleRemote(_ context.Context, name string) (*models.Remote, error) {
|
||||
return &models.Remote{Name: name, RepoType: models.RepoTypeRemote, ServeStale: true}, nil
|
||||
}
|
||||
|
||||
func TestServeStaleDebAndAlpine(t *testing.T) {
|
||||
var down atomic.Bool
|
||||
e := fakeEngine(nil)
|
||||
e.getRemote = staleRemote
|
||||
e.debMember = func(_ context.Context, name, _ string, _ bool) (*DebMember, error) {
|
||||
if down.Load() {
|
||||
return nil, errors.New("upstream down")
|
||||
}
|
||||
return suiteDebMember(name, "1"), nil
|
||||
}
|
||||
e.apkMember = func(_ context.Context, name, _ string) (*AlpineMember, error) {
|
||||
if down.Load() {
|
||||
return nil, errors.New("upstream down")
|
||||
}
|
||||
return &AlpineMember{RemoteName: name, Index: []byte(apkRec("foo", "1-r0", "x86_64"))}, nil
|
||||
}
|
||||
for _, tc := range []struct {
|
||||
virt models.Virtual
|
||||
path string
|
||||
}{
|
||||
{debVirt("a"), "dists/s/Release"},
|
||||
{alpineVirt("a"), "x86_64/APKINDEX.tar.gz"},
|
||||
} {
|
||||
down.Store(false)
|
||||
first, _, err := e.Fetch(context.Background(), tc.virt, tc.path, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
down.Store(true)
|
||||
got, _, err := e.Fetch(context.Background(), tc.virt, tc.path, "")
|
||||
if err != nil || string(got) != string(first) {
|
||||
t.Fatalf("%s: want the previous copy while the member is down, got err %v", tc.virt.PackageType, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestServeStaleEvictionDropsLateRefresh(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
var slow atomic.Bool
|
||||
e := fakeEngine(nil)
|
||||
e.mergeTTL = time.Minute
|
||||
e.getRemote = staleRemote
|
||||
e.debMember = func(_ context.Context, name, base string, _ bool) (*DebMember, error) {
|
||||
if base == "dists/s0/" && slow.Load() {
|
||||
<-release
|
||||
}
|
||||
return suiteDebMember(name, "1"), nil
|
||||
}
|
||||
fetch := func(suite string) {
|
||||
t.Helper()
|
||||
if _, _, err := e.Fetch(context.Background(), debVirt("a"), "dists/"+suite+"/Release", ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
fetch("s0")
|
||||
slow.Store(true)
|
||||
e.mu.Lock()
|
||||
e.gens["deb\x00v\x00dists/s0/"].at = time.Time{}
|
||||
e.mu.Unlock()
|
||||
fetch("s0")
|
||||
for i := range maxDebSuites {
|
||||
time.Sleep(time.Microsecond)
|
||||
fetch(fmt.Sprintf("s%d", i+1))
|
||||
}
|
||||
close(release)
|
||||
sk := "deb\x00v\x00dists/s0/\x00a"
|
||||
waitRefresh(e, sk)
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
if e.gens["deb\x00v\x00dists/s0/"] != nil {
|
||||
t.Fatal("suite s0 should be evicted")
|
||||
}
|
||||
if e.stale[sk] != nil {
|
||||
t.Fatal("a refresh finishing after eviction must not re-insert its copy")
|
||||
}
|
||||
if len(e.stale) != maxDebSuites {
|
||||
t.Fatalf("stale holds %d copies, want %d", len(e.stale), maxDebSuites)
|
||||
}
|
||||
}
|
||||
|
||||
func TestServeStaleEvictionDropsLateColdFetch(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
var slow atomic.Bool
|
||||
e := fakeEngine(nil)
|
||||
e.mergeTTL = time.Minute
|
||||
e.getRemote = staleRemote
|
||||
e.debMember = func(_ context.Context, name, base string, _ bool) (*DebMember, error) {
|
||||
if base == "dists/s0/" && slow.Load() {
|
||||
<-release
|
||||
}
|
||||
return suiteDebMember(name, "1"), nil
|
||||
}
|
||||
fetch := func(suite string) {
|
||||
if _, _, err := e.Fetch(context.Background(), debVirt("a"), "dists/"+suite+"/Release", ""); err != nil {
|
||||
t.Error(err)
|
||||
}
|
||||
}
|
||||
sk := "deb\x00v\x00dists/s0/\x00a"
|
||||
fetch("s0")
|
||||
slow.Store(true)
|
||||
e.mu.Lock()
|
||||
e.gens["deb\x00v\x00dists/s0/"].at = time.Time{}
|
||||
delete(e.stale, sk)
|
||||
e.mu.Unlock()
|
||||
done := make(chan struct{})
|
||||
go func() { defer close(done); fetch("s0") }()
|
||||
for {
|
||||
e.mu.Lock()
|
||||
cold := e.stale[sk] != nil
|
||||
e.mu.Unlock()
|
||||
if cold {
|
||||
break
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
for i := range maxDebSuites {
|
||||
time.Sleep(time.Microsecond)
|
||||
fetch(fmt.Sprintf("s%d", i+1))
|
||||
}
|
||||
close(release)
|
||||
<-done
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
if e.stale[sk] != nil {
|
||||
t.Fatal("a cold fetch finishing after eviction must not re-insert its copy")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergePrunesExpiredCopiesOfOtherMembers(t *testing.T) {
|
||||
e, _ := staleEngine(false, func() *RPMMember { return rpmPkgMember("a", "1") })
|
||||
e.stale = map[string]*staleEntry{"rpm\x00gone\x00x": {m: rpmPkgMember("x", "1"), at: time.Now().Add(-maxStaleAge)}}
|
||||
fetchRepomd(t, e)
|
||||
if len(e.stale) != 0 {
|
||||
t.Fatal("any merge should prune expired copies of removed members")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
-- serve_stale lets a virtual merge a member's previous index while that member
|
||||
-- refreshes in the background.
|
||||
ALTER TABLE remotes ADD COLUMN IF NOT EXISTS serve_stale BOOLEAN NOT NULL DEFAULT FALSE;
|
||||
@@ -77,6 +77,9 @@ type Remote struct {
|
||||
QuarantineDays int `json:"quarantine_days,omitempty"`
|
||||
|
||||
StaleOnError bool `json:"stale_on_error"`
|
||||
// ServeStale lets an rpm/deb/alpine virtual merge this member's previous
|
||||
// index while it refreshes in the background.
|
||||
ServeStale bool `json:"serve_stale"`
|
||||
|
||||
ReleasesRemote string `json:"releases_remote,omitempty"`
|
||||
ManagedBy string `json:"managed_by,omitempty"`
|
||||
|
||||
@@ -18,6 +18,7 @@ export interface Remote {
|
||||
quarantine_enabled: boolean;
|
||||
quarantine_days: number;
|
||||
stale_on_error: boolean;
|
||||
serve_stale: boolean;
|
||||
releases_remote: string;
|
||||
managed_by: string;
|
||||
created_at: string;
|
||||
|
||||
@@ -50,6 +50,8 @@ export function RemoteDetail() {
|
||||
<dd>{remote.check_mutable ? 'enabled' : 'disabled'}</dd>
|
||||
<dt>Stale on Error</dt>
|
||||
<dd>{remote.stale_on_error ? 'enabled' : 'disabled'}</dd>
|
||||
<dt>Serve Stale</dt>
|
||||
<dd>{remote.serve_stale ? 'enabled' : 'disabled'}</dd>
|
||||
{remote.releases_remote && (
|
||||
<>
|
||||
<dt>Releases Remote</dt>
|
||||
|
||||
Reference in New Issue
Block a user