mirror of
https://github.com/git-pkgs/proxy.git
synced 2026-11-09 15:49:16 -05:00
315 lines
12 KiB
Go
315 lines
12 KiB
Go
package handler
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
"net/http"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/git-pkgs/proxy/internal/storage"
|
|
)
|
|
|
|
// archHookStorage wraps a storage with hooks to force interleavings.
|
|
type archHookStorage struct {
|
|
storage.Storage
|
|
failStore func(path string) error
|
|
afterStore func(path string)
|
|
beforeDelete func(path string)
|
|
}
|
|
|
|
func (s *archHookStorage) Store(ctx context.Context, path string, r io.Reader) (int64, string, error) {
|
|
if s.failStore != nil {
|
|
if err := s.failStore(path); err != nil {
|
|
return 0, "", err
|
|
}
|
|
}
|
|
n, hash, err := s.Storage.Store(ctx, path, r)
|
|
if s.afterStore != nil {
|
|
s.afterStore(path)
|
|
}
|
|
return n, hash, err
|
|
}
|
|
|
|
func (s *archHookStorage) Delete(ctx context.Context, path string) error {
|
|
if s.beforeDelete != nil {
|
|
s.beforeDelete(path)
|
|
}
|
|
return s.Storage.Delete(ctx, path)
|
|
}
|
|
|
|
const archTestDB, archTestSig = "/custom/os/x86_64/custom.db", "/custom/os/x86_64/custom.db.sig"
|
|
|
|
// pauseOnce returns a hook that stops the first call for a path starting with
|
|
// prefix until release
|
|
// closes or a short timeout passes, so a test does not hang when locking
|
|
// keeps the other writer out.
|
|
func pauseOnce(prefix string, paused, release chan struct{}) func(string) {
|
|
var fired atomic.Bool
|
|
return func(p string) {
|
|
if !strings.HasPrefix(p, prefix) || !fired.CompareAndSwap(false, true) {
|
|
return
|
|
}
|
|
close(paused)
|
|
select {
|
|
case <-release:
|
|
case <-time.After(200 * time.Millisecond):
|
|
}
|
|
}
|
|
}
|
|
|
|
// waitPaused waits until a pauseOnce hook fired.
|
|
func waitPaused(t *testing.T, paused <-chan struct{}) {
|
|
t.Helper()
|
|
select {
|
|
case <-paused:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("hook never fired")
|
|
}
|
|
}
|
|
|
|
// waitBriefly waits for done, or gives up after a moment when a lock holds the
|
|
// call back.
|
|
func waitBriefly(done <-chan struct{}) {
|
|
select {
|
|
case <-done:
|
|
case <-time.After(300 * time.Millisecond):
|
|
}
|
|
}
|
|
|
|
// TestArchHandler_ReplicasWritingOutOfOrder covers two proxy replicas sharing
|
|
// database and storage, whose database writes finish out of order: the cache
|
|
// row must never describe other bytes than its blob, or every client gets the
|
|
// newest database with an older signature.
|
|
func TestArchHandler_ReplicasWritingOutOfOrder(t *testing.T) {
|
|
upstream := newArchUpstream(t, map[string][]byte{archTestDB: []byte("db v1"), archTestSig: []byte("sig v1")})
|
|
upstream.broken[archTestSig] = true
|
|
proxy := newArchTestProxy(t, upstream)
|
|
replicaA := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL})
|
|
replicaB := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL})
|
|
d := replicaA.database("custom", upstream.URL, archTestDB[1:])
|
|
|
|
assertArchBody(t, serveArchRequest(replicaA, http.MethodGet, "/custom"+archTestDB, nil), "db v1", []byte("db v1"))
|
|
upstream.mu.Lock()
|
|
upstream.files[archTestDB], upstream.files[archTestSig] = []byte("db v2"), []byte("sig v2")
|
|
upstream.broken[archTestSig] = false
|
|
upstream.mu.Unlock()
|
|
ageMetadata(t, proxy, replicaA, "custom", upstream.URL, archTestDB[1:])
|
|
|
|
// Replica A's signature recovery stores db v2 and stalls before its row
|
|
// update.
|
|
paused, release := make(chan struct{}), make(chan struct{})
|
|
proxy.Storage = &archHookStorage{Storage: proxy.Storage,
|
|
afterStore: pauseOnce("_metadata/"+archEcosystem+"/"+d.dbKey+"/", paused, release)}
|
|
recovered := make(chan struct{})
|
|
go func() {
|
|
serveArchRequest(replicaA, http.MethodGet, "/custom"+archTestSig, nil)
|
|
close(recovered)
|
|
}()
|
|
waitPaused(t, paused)
|
|
|
|
// Replica B pairs v2, then caches v3 whose signature fetch fails.
|
|
assertArchBody(t, serveArchRequest(replicaB, http.MethodGet, "/custom"+archTestDB, nil), "B db v2", []byte("db v2"))
|
|
assertArchBody(t, serveArchRequest(replicaB, http.MethodGet, "/custom"+archTestSig, nil), "B sig v2", []byte("sig v2"))
|
|
upstream.mu.Lock()
|
|
upstream.files[archTestDB], upstream.files[archTestSig] = []byte("db v3"), []byte("sig v3")
|
|
upstream.broken[archTestSig] = true
|
|
upstream.mu.Unlock()
|
|
ageMetadata(t, proxy, replicaB, "custom", upstream.URL, archTestDB[1:])
|
|
assertArchBody(t, serveArchRequest(replicaB, http.MethodGet, "/custom"+archTestDB, nil), "B db v3", []byte("db v3"))
|
|
|
|
close(release)
|
|
<-recovered
|
|
|
|
db := serveArchRequest(replicaB, http.MethodGet, "/custom"+archTestDB, nil)
|
|
sig := serveArchRequest(replicaB, http.MethodGet, "/custom"+archTestSig, nil)
|
|
if sig.Code == http.StatusOK && strings.Replace(db.Body.String(), "db", "sig", 1) != sig.Body.String() {
|
|
t.Errorf("%q served with %q", db.Body.String(), sig.Body.String())
|
|
}
|
|
}
|
|
|
|
// TestArchHandler_FailedDatabaseWriteKeepsPair covers a database refresh whose
|
|
// cache write fails: the proxy must not hand out the new database while the
|
|
// cache still pairs the old one with its signature.
|
|
func TestArchHandler_FailedDatabaseWriteKeepsPair(t *testing.T) {
|
|
upstream := newArchUpstream(t, map[string][]byte{archTestDB: []byte("db v1"), archTestSig: []byte("sig v1")})
|
|
proxy := newArchTestProxy(t, upstream)
|
|
h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL})
|
|
d := h.database("custom", upstream.URL, archTestDB[1:])
|
|
var fail atomic.Bool
|
|
proxy.Storage = &archHookStorage{Storage: proxy.Storage, failStore: func(path string) error {
|
|
if fail.Load() && strings.HasPrefix(path, archBlobPath(d.dbKey, "")) {
|
|
return errors.New("disk full")
|
|
}
|
|
return nil
|
|
}}
|
|
|
|
assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil), "db v1", []byte("db v1"))
|
|
assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil), "sig v1", []byte("sig v1"))
|
|
upstream.set(archTestDB, []byte("db v2"))
|
|
upstream.set(archTestSig, []byte("sig v2"))
|
|
ageMetadata(t, proxy, h, "custom", upstream.URL, archTestDB[1:])
|
|
fail.Store(true)
|
|
|
|
db := serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil)
|
|
sig := serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil)
|
|
if db.Code == http.StatusOK && sig.Code == http.StatusOK && strings.Replace(db.Body.String(), "db", "sig", 1) != sig.Body.String() {
|
|
t.Errorf("%q served with %q", db.Body.String(), sig.Body.String())
|
|
}
|
|
}
|
|
|
|
// TestArchHandler_ConcurrentSignatureWritesLeaveOneBlob covers two signature
|
|
// writes overlapping: afterwards exactly one signature blob is left and the
|
|
// cache row points at it.
|
|
func TestArchHandler_ConcurrentSignatureWritesLeaveOneBlob(t *testing.T) {
|
|
d1, d2, d3 := archDigest([]byte("1")), archDigest([]byte("2")), archDigest([]byte("3"))
|
|
|
|
for _, tt := range []struct {
|
|
name string
|
|
hook func(h *ArchHandler, d archDatabase, s *archHookStorage, paused, release chan struct{})
|
|
first string // digest stored by the stalled write
|
|
other string // digest stored by the write overtaking it
|
|
}{
|
|
{"stalled after store", func(_ *ArchHandler, d archDatabase, s *archHookStorage, paused, release chan struct{}) {
|
|
s.afterStore = pauseOnce(archSignaturePath(d, d2), paused, release)
|
|
}, d2, d3},
|
|
{"stalled before delete", func(_ *ArchHandler, d archDatabase, s *archHookStorage, paused, release chan struct{}) {
|
|
s.beforeDelete = pauseOnce(archSignaturePath(d, d1), paused, release)
|
|
}, d2, d1},
|
|
} {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
upstream := newArchUpstream(t, map[string][]byte{})
|
|
proxy := newArchTestProxy(t, upstream)
|
|
mock := proxy.Storage.(*mockStorage)
|
|
h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL})
|
|
d := h.database("custom", upstream.URL, archTestDB[1:])
|
|
ctx := context.Background()
|
|
h.storeSignature(ctx, d, d1, &upstreamMetadata{body: []byte("sig 1")})
|
|
|
|
hooks := &archHookStorage{Storage: mock}
|
|
proxy.Storage = hooks
|
|
paused, release := make(chan struct{}), make(chan struct{})
|
|
tt.hook(h, d, hooks, paused, release)
|
|
|
|
stalled := make(chan struct{})
|
|
go func() {
|
|
h.storeSignature(ctx, d, tt.first, &upstreamMetadata{body: []byte("sig first")})
|
|
close(stalled)
|
|
}()
|
|
waitPaused(t, paused)
|
|
overtaking := make(chan struct{})
|
|
go func() {
|
|
h.storeSignature(ctx, d, tt.other, &upstreamMetadata{body: []byte("sig other")})
|
|
close(overtaking)
|
|
}()
|
|
waitBriefly(overtaking)
|
|
close(release)
|
|
<-stalled
|
|
<-overtaking
|
|
|
|
row, _ := proxy.DB.GetMetadataCache(archEcosystem, d.sigKey)
|
|
if _, err := mock.Open(ctx, row.StoragePath); err != nil {
|
|
t.Errorf("row points at %s, which is gone: %v", row.StoragePath, err)
|
|
}
|
|
mock.mu.Lock()
|
|
defer mock.mu.Unlock()
|
|
var blobs []string
|
|
for path := range mock.files {
|
|
if strings.HasPrefix(path, archBlobPath(d.sigKey, "")) {
|
|
blobs = append(blobs, path)
|
|
}
|
|
}
|
|
if len(blobs) != 1 {
|
|
t.Errorf("signature blobs = %v, want only the row's", blobs)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestArchHandler_MissingSignatureRecheckedAfterTTL covers a 404 for the
|
|
// signature that was only transient: once the TTL is over the proxy asks
|
|
// again instead of answering 404 until the next database is published.
|
|
func TestArchHandler_MissingSignatureRecheckedAfterTTL(t *testing.T) {
|
|
upstream := newArchUpstream(t, map[string][]byte{archTestDB: []byte("db v1")})
|
|
proxy := newArchTestProxy(t, upstream)
|
|
h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL})
|
|
|
|
assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil), "db", []byte("db v1"))
|
|
if w := serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil); w.Code != http.StatusNotFound {
|
|
t.Fatalf("signature: status = %d, want 404", w.Code)
|
|
}
|
|
|
|
upstream.set(archTestSig, []byte("sig v1"))
|
|
ageMetadata(t, proxy, h, "custom", upstream.URL, archTestDB[1:])
|
|
ageMetadata(t, proxy, h, "custom", upstream.URL, archTestSig[1:])
|
|
|
|
assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil), "db refreshed", []byte("db v1"))
|
|
assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil), "signature after TTL", []byte("sig v1"))
|
|
}
|
|
|
|
// TestArchHandler_SlowFetchDoesNotOverwriteNewerDatabase covers a signature
|
|
// recovery whose database fetch is still in flight while other clients
|
|
// refresh to newer databases. Its result must not replace the newer cached
|
|
// database, or a client holding that one gets an older signature.
|
|
func TestArchHandler_SlowFetchDoesNotOverwriteNewerDatabase(t *testing.T) {
|
|
upstream := newArchUpstream(t, map[string][]byte{archTestDB: []byte("db v1"), archTestSig: []byte("sig v1")})
|
|
upstream.broken[archTestSig] = true
|
|
proxy := newArchTestProxy(t, upstream)
|
|
h := NewArchHandler(proxy, "http://proxy.example", map[string]string{"custom": upstream.URL})
|
|
|
|
assertArchBody(t, serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil), "db v1", []byte("db v1"))
|
|
upstream.mu.Lock()
|
|
upstream.files[archTestDB], upstream.files[archTestSig] = []byte("db v2"), []byte("sig v2")
|
|
upstream.broken[archTestSig] = false
|
|
upstream.mu.Unlock()
|
|
ageMetadata(t, proxy, h, "custom", upstream.URL, archTestDB[1:])
|
|
|
|
// Client A's signature recovery has fetched db v2; the response is slow.
|
|
paused, release := make(chan struct{}), make(chan struct{})
|
|
upstream.mu.Lock()
|
|
upstream.served = pauseOnce(archTestDB, paused, release)
|
|
upstream.mu.Unlock()
|
|
recovered := make(chan struct{})
|
|
go func() {
|
|
serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil)
|
|
close(recovered)
|
|
}()
|
|
waitPaused(t, paused)
|
|
|
|
// Client B refreshes meanwhile, then the upstream moves to v3, whose
|
|
// signature fails, and client C refreshes to it.
|
|
b := make(chan struct{})
|
|
go func() {
|
|
serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil)
|
|
serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil)
|
|
close(b)
|
|
}()
|
|
waitBriefly(b)
|
|
upstream.mu.Lock()
|
|
upstream.files[archTestDB], upstream.files[archTestSig] = []byte("db v3"), []byte("sig v3")
|
|
upstream.broken[archTestSig] = true
|
|
upstream.mu.Unlock()
|
|
ageMetadata(t, proxy, h, "custom", upstream.URL, archTestDB[1:])
|
|
var cDB string
|
|
c := make(chan struct{})
|
|
go func() {
|
|
cDB = serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil).Body.String()
|
|
close(c)
|
|
}()
|
|
waitBriefly(c)
|
|
|
|
close(release)
|
|
<-recovered
|
|
<-b
|
|
<-c
|
|
|
|
if sig := serveArchRequest(h, http.MethodGet, "/custom"+archTestSig, nil); sig.Code == http.StatusOK && strings.Replace(cDB, "db", "sig", 1) != sig.Body.String() {
|
|
t.Errorf("client holding %q got %q", cDB, sig.Body.String())
|
|
}
|
|
if got := serveArchRequest(h, http.MethodGet, "/custom"+archTestDB, nil).Body.String(); got < cDB {
|
|
t.Errorf("cache went back to %q after serving %q", got, cDB)
|
|
}
|
|
}
|