unkin-agent
327f09fc6f
Add a serve_stale remote flag. rpm, deb and alpine virtual merges use a flagged member's last good fetch immediately and refresh it in the background, single-flighted per member.
114 lines
2.8 KiB
Go
114 lines
2.8 KiB
Go
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)
|
|
}
|