diff --git a/docs/common/caching.md b/docs/common/caching.md index fcbe5fd..b49b16c 100644 --- a/docs/common/caching.md +++ b/docs/common/caching.md @@ -15,7 +15,9 @@ 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. 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`. | + +`PUT` replaces the whole remote: any field omitted from the body, such as `stale_on_error` or `serve_stale`, resets to its default (`mutable_ttl` is the exception above). | `quarantine_enabled` / `quarantine_days` | Stored on the remote but not enforced by the proxy yet. | | `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. | diff --git a/internal/server/server_test.go b/internal/server/server_test.go index fa12f6c..14d4c15 100644 --- a/internal/server/server_test.go +++ b/internal/server/server_test.go @@ -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"}`) diff --git a/internal/tui/render.go b/internal/tui/render.go index 67d7298..635a8ca 100644 --- a/internal/tui/render.go +++ b/internal/tui/render.go @@ -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, ", "))) diff --git a/internal/virtual/engine.go b/internal/virtual/engine.go index 808880a..21225b6 100644 --- a/internal/virtual/engine.go +++ b/internal/virtual/engine.go @@ -57,9 +57,18 @@ type Engine struct { 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]any + 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 @@ -535,30 +544,41 @@ func (e *Engine) fetchAlpineMember(ctx context.Context, name, dir string) (*Alpi } // staleMember fetches member name for the merge under key. A serve_stale member -// with a previous good fetch returns it at once and refreshes in the -// background, one refresh per member at a time; the next merge picks it up. +// 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) } - sk := key + "\x00" + name + e.mu.Lock() + for k, se := range e.stale { + if time.Since(se.at) >= maxStaleAge { + delete(e.stale, k) + } + } + prev := e.stale[sk] + e.mu.Unlock() refresh := func() (any, error) { m, err := fetch(ctx) if err == nil { e.mu.Lock() - if e.stale == nil { - e.stale = map[string]any{} + // An evicted or replaced entry means this refresh is no longer wanted. + if e.stale[sk] == prev { + if e.stale == nil { + e.stale = map[string]*staleEntry{} + } + e.stale[sk] = &staleEntry{m: m, at: time.Now()} } - e.stale[sk] = m e.mu.Unlock() } return m, err } - e.mu.Lock() - prev, ok := e.stale[sk].(*T) - e.mu.Unlock() - if !ok { + if prev == nil { m, err := refresh() return m.(*T), err } @@ -568,7 +588,7 @@ func staleMember[T any](ctx context.Context, e *Engine, key, name string, fetch slog.Warn("virtual member background refresh failed", "member", name, "error", r.Err) } }() - return prev, nil + return prev.m.(*T), nil } // evictLRU drops the least recently used merges under prefix beyond keep. diff --git a/internal/virtual/engine_stale_test.go b/internal/virtual/engine_stale_test.go index acb7657..62e4c0c 100644 --- a/internal/virtual/engine_stale_test.go +++ b/internal/virtual/engine_stale_test.go @@ -2,6 +2,8 @@ package virtual import ( "context" + "errors" + "fmt" "sync/atomic" "testing" "time" @@ -16,18 +18,32 @@ func rpmPkgMember(name, ver string) *RPMMember { // 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: stale}, nil + return &models.Remote{Name: name, RepoType: models.RepoTypeRemote, ServeStale: flag.Load()}, nil } e.rpmMember = func(_ context.Context, name string) (*RPMMember, error) { calls.Add(1) - return fetch(), nil + return fetch() } - return e, &calls + 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", "") @@ -111,3 +127,147 @@ func TestServeStaleColdStartBlocksMerge(t *testing.T) { }) 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) (*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) (*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) + } +} diff --git a/ui/src/api/types.ts b/ui/src/api/types.ts index e3efbd8..ab5fe12 100644 --- a/ui/src/api/types.ts +++ b/ui/src/api/types.ts @@ -17,6 +17,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; diff --git a/ui/src/pages/RemoteDetail.tsx b/ui/src/pages/RemoteDetail.tsx index f8a5063..96cb03b 100644 --- a/ui/src/pages/RemoteDetail.tsx +++ b/ui/src/pages/RemoteDetail.tsx @@ -50,6 +50,8 @@ export function RemoteDetail() {