diff --git a/docs/common/caching.md b/docs/common/caching.md index 3b32604..fcbe5fd 100644 --- a/docs/common/caching.md +++ b/docs/common/caching.md @@ -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. Default `false`. | | `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/database/remotes.go b/internal/database/remotes.go index 44b2eb5..48ae1e2 100644 --- a/internal/database/remotes.go +++ b/internal/database/remotes.go @@ -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 - ) 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) + mirror_strategy, 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) `, 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), + normalizeMirrorStrategy(r.MirrorStrategy), 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, + mirror_strategy=$26, serve_stale=$27, 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), + normalizeMirrorStrategy(r.MirrorStrategy), r.ServeStale, ) return err } diff --git a/internal/virtual/engine.go b/internal/virtual/engine.go index 9ab1905..808880a 100644 --- a/internal/virtual/engine.go +++ b/internal/virtual/engine.go @@ -55,6 +55,9 @@ 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]any } // mergedRepo is a merged index document plus the files it references. @@ -353,7 +356,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) }) + 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) }) + }) if err != nil { return nil, err } @@ -402,7 +407,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) }) + 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) }) + }) if err != nil { return nil, err } @@ -453,8 +460,11 @@ func (e *Engine) fetchDebMember(ctx context.Context, name, base string) (*DebMem } 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 } @@ -524,6 +534,43 @@ 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 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. +func staleMember[T any](ctx context.Context, e *Engine, key, name string, fetch func(context.Context) (*T, error)) (*T, error) { + remote, err := e.getRemote(ctx, name) + if err != nil || !remote.ServeStale { + return fetch(ctx) + } + sk := key + "\x00" + name + refresh := func() (any, error) { + m, err := fetch(ctx) + if err == nil { + e.mu.Lock() + if e.stale == nil { + e.stale = map[string]any{} + } + e.stale[sk] = m + e.mu.Unlock() + } + return m, err + } + e.mu.Lock() + prev, ok := e.stale[sk].(*T) + e.mu.Unlock() + if !ok { + 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, nil +} + // evictLRU drops the least recently used merges under prefix beyond keep. func (e *Engine) evictLRU(prefix string, keep int) { e.mu.Lock() @@ -537,6 +584,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) + } + } } } diff --git a/internal/virtual/engine_stale_test.go b/internal/virtual/engine_stale_test.go new file mode 100644 index 0000000..acb7657 --- /dev/null +++ b/internal/virtual/engine_stale_test.go @@ -0,0 +1,113 @@ +package virtual + +import ( + "context" + "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) { + var calls atomic.Int32 + e := fakeEngine(nil) + e.getRemote = func(_ context.Context, name string) (*models.Remote, error) { + return &models.Remote{Name: name, RepoType: models.RepoTypeRemote, ServeStale: stale}, nil + } + e.rpmMember = func(_ context.Context, name string) (*RPMMember, error) { + calls.Add(1) + return fetch(), nil + } + return e, &calls +} + +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) +} diff --git a/migrations/0008_remote_serve_stale.sql b/migrations/0008_remote_serve_stale.sql new file mode 100644 index 0000000..1434394 --- /dev/null +++ b/migrations/0008_remote_serve_stale.sql @@ -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; diff --git a/pkg/models/remote.go b/pkg/models/remote.go index fd27aa6..538a875 100644 --- a/pkg/models/remote.go +++ b/pkg/models/remote.go @@ -73,6 +73,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"`