Bound serve_stale member copies by age, toggle and eviction
ci/woodpecker/pr/pre-commit Pipeline was successful
ci/woodpecker/pr/test Pipeline was successful
ci/woodpecker/pr/build Pipeline was successful

- drop a member's copy when serve_stale is off
- block on the member again once its copy is over an hour old
- skip storing a refresh whose entry was evicted or replaced
- show serve_stale in the UI and TUI
This commit is contained in:
2026-10-25 13:29:43 +11:00
parent 327f09fc6f
commit 50ef1291fc
7 changed files with 221 additions and 16 deletions
+3 -1
View File
@@ -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. |
+19
View File
@@ -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"}`)
+1
View File
@@ -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, ", ")))
+32 -12
View File
@@ -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.
+163 -3
View File
@@ -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)
}
}
+1
View File
@@ -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;
+2
View File
@@ -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>