diff --git a/internal/api/v2/objects_evict_test.go b/internal/api/v2/objects_evict_test.go index 872c8a4..75f38c2 100644 --- a/internal/api/v2/objects_evict_test.go +++ b/internal/api/v2/objects_evict_test.go @@ -2,6 +2,8 @@ package v2 import ( "context" + "errors" + "fmt" "net/http" "net/http/httptest" "testing" @@ -40,9 +42,18 @@ func TestRemoteEvictDelegatesToEvictor(t *testing.T) { } } -func TestRemoteEvictMapsProxyErrorStatus(t *testing.T) { - ev := &fakeEvictor{err: &proxy.ProxyError{Status: http.StatusNotFound, Message: "remote not found"}} - if code := deleteObject(ev, "x"); code != http.StatusNotFound { - t.Errorf("code = %d, want 404", code) +func TestRemoteEvictMapsErrorStatus(t *testing.T) { + for name, tc := range map[string]struct { + err error + want int + }{ + "bad wildcard": {&proxy.ProxyError{Status: http.StatusBadRequest, Message: "wildcard evict must be /*"}, http.StatusBadRequest}, + "unknown remote": {&proxy.ProxyError{Status: http.StatusNotFound, Message: "remote not found"}, http.StatusNotFound}, + "lock busy": {fmt.Errorf("wrapped: %w", &proxy.ProxyError{Status: http.StatusServiceUnavailable, Message: "retry"}), http.StatusServiceUnavailable}, + "other error": {errors.New("db down"), http.StatusInternalServerError}, + } { + if code := deleteObject(&fakeEvictor{err: tc.err}, "x"); code != tc.want { + t.Errorf("%s: code = %d, want %d", name, code, tc.want) + } } } diff --git a/internal/proxy/engine.go b/internal/proxy/engine.go index e760962..45a1b87 100644 --- a/internal/proxy/engine.go +++ b/internal/proxy/engine.go @@ -49,6 +49,8 @@ type Engine struct { // mirror strategy to prefer the mirror currently handling the fewest // requests. Per-replica and approximate, which is fine. inflight sync.Map + // evictLockWait bounds how long Evict waits on a held fetch lock. + evictLockWait time.Duration } func NewEngine(db *database.DB, c *cache.Redis, s *storage.S3) *Engine { @@ -59,6 +61,8 @@ func NewEngine(db *database.DB, c *cache.Redis, s *storage.S3) *Engine { cas: storage.NewCAS(s), circuit: NewCircuitBreaker(c), accessLog: make(chan database.AccessLogEntry, accessLogBufferSize), + + evictLockWait: fetchLockTTL, } go e.runAccessLogWriter() return e @@ -212,11 +216,11 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p // Evict drops path from every cache layer (artifact row, index object, Redis // freshness and ETag keys) so the next request refetches from upstream. A -// trailing "*" evicts every path under that prefix; a bare "*" is rejected. +// trailing "/*" evicts every path under that directory. func (e *Engine) Evict(ctx context.Context, remoteName, path string) error { prefix, wildcard := strings.CutSuffix(path, "*") - if wildcard && prefix == "" { - return &ProxyError{Status: http.StatusBadRequest, Message: "refusing to evict an entire remote"} + if wildcard && !strings.HasSuffix(prefix, "/") { + return &ProxyError{Status: http.StatusBadRequest, Message: "wildcard evict must be /*"} } if _, err := e.db.GetRemote(ctx, remoteName); errors.Is(err, pgx.ErrNoRows) { return &ProxyError{Status: http.StatusNotFound, Message: fmt.Sprintf("remote %q not found", remoteName)} @@ -224,9 +228,10 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error { return fmt.Errorf("get remote: %w", err) } if !wildcard { - if e.waitForLock(ctx, remoteName, path) { - defer e.cache.ReleaseLock(ctx, remoteName, path) + if err := e.waitForLock(ctx, remoteName, path); err != nil { + return err } + defer func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path) }() if err := e.db.DeleteArtifact(ctx, remoteName, path); err != nil { return fmt.Errorf("delete artifact: %w", err) } @@ -235,6 +240,9 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error { } return e.cache.ForgetPath(ctx, remoteName, path) } + // ponytail: no lock for wildcards; a Fetch already in flight under the + // prefix can re-cache its path after the evict. Per-path locks over a + // directory would close it if that ever matters. if err := e.db.DeleteArtifactsByPrefix(ctx, remoteName, prefix); err != nil { return fmt.Errorf("delete artifacts: %w", err) } @@ -245,17 +253,27 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error { } // waitForLock takes the per-path fetch lock so an in-flight Fetch cannot -// re-set TTL/ETag keys after an evict. It gives up once the lock would have -// expired anyway, or when ctx ends. -func (e *Engine) waitForLock(ctx context.Context, remoteName, path string) bool { - deadline := time.Now().Add(fetchLockTTL) +// re-set TTL/ETag keys after an evict. It fails with a 503 when the lock +// cannot be taken within evictLockWait or Redis errors. +func (e *Engine) waitForLock(ctx context.Context, remoteName, path string) error { + deadline := time.Now().Add(e.evictLockWait) for { - if ok, err := e.cache.AcquireLock(ctx, remoteName, path, fetchLockTTL); ok || err != nil || time.Now().After(deadline) { - return ok + ok, err := e.cache.AcquireLock(ctx, remoteName, path, fetchLockTTL) + if ok { + return nil + } + if ctx.Err() != nil { + return ctx.Err() + } + if err != nil { + return &ProxyError{Status: http.StatusServiceUnavailable, Message: fmt.Sprintf("fetch lock: %v", err)} + } + if time.Now().After(deadline) { + return &ProxyError{Status: http.StatusServiceUnavailable, Message: "fetch in progress, retry evict"} } select { case <-ctx.Done(): - return false + return ctx.Err() case <-time.After(50 * time.Millisecond): } } diff --git a/internal/proxy/evict_test.go b/internal/proxy/evict_test.go index c27e47a..15b8def 100644 --- a/internal/proxy/evict_test.go +++ b/internal/proxy/evict_test.go @@ -10,6 +10,7 @@ import ( "time" _ "git.unkin.net/unkin/artifactapi/internal/provider/rpm" + "git.unkin.net/unkin/artifactapi/internal/storage" "git.unkin.net/unkin/artifactapi/pkg/models" ) @@ -51,6 +52,32 @@ func rpmRemote(t *testing.T, name, baseURL string) models.Remote { return seed(t, models.Remote{Name: name, PackageType: models.PackageRPM, RepoType: models.RepoTypeRemote, BaseURL: baseURL, MutableTTL: 7200, CheckMutable: true}) } +// cached reports which cache layers hold path: artifact row, index object, +// Redis TTL key, Redis ETag key. +type cached struct{ row, index, ttl, etag bool } + +// Mutable indexes live in the S3 index; immutable blobs get an artifact row. +var ( + indexCached = cached{index: true, ttl: true, etag: true} + blobCached = cached{row: true, ttl: true, etag: true} +) + +func layers(t *testing.T, remote, path string) cached { + t.Helper() + ctx := context.Background() + var c cached + _, err := testDB.GetArtifact(ctx, remote, path) + c.row = err == nil + c.index, err = testEngine.store.Exists(ctx, storage.IndexKey(remote, path)) + if err != nil { + t.Fatalf("stat index %s: %v", path, err) + } + c.ttl, _ = testCache.CheckTTL(ctx, remote, path) + etag, _ := testCache.GetETag(ctx, remote, path) + c.etag = etag != "" + return c +} + func TestEvictMutableIndexRefetches(t *testing.T) { requireStack(t) srv, rev, conditional := changingUpstream(t) @@ -60,6 +87,9 @@ func TestEvictMutableIndexRefetches(t *testing.T) { if got := fetchBody(t, r, path); got != "rev1:/"+path { t.Fatalf("initial fetch = %q", got) } + if c := layers(t, r.Name, path); c != indexCached { + t.Fatalf("before evict = %+v, want %+v", c, indexCached) + } rev.Store("rev2") if got := fetchBody(t, r, path); got != "rev1:/"+path { t.Fatalf("within TTL = %q, want cached rev1", got) @@ -67,6 +97,9 @@ func TestEvictMutableIndexRefetches(t *testing.T) { if err := testEngine.Evict(context.Background(), r.Name, path); err != nil { t.Fatalf("evict: %v", err) } + if c := layers(t, r.Name, path); c != (cached{}) { + t.Fatalf("after evict = %+v, want every layer gone", c) + } conditional.Store(0) if got := fetchBody(t, r, path); got != "rev2:/"+path { t.Fatalf("after evict = %q, want rev2", got) @@ -76,19 +109,38 @@ func TestEvictMutableIndexRefetches(t *testing.T) { } } -func TestEvictRejectsBareWildcard(t *testing.T) { +func TestEvictImmutableBlobDropsRow(t *testing.T) { + requireStack(t) + srv, _, _ := changingUpstream(t) + r := rpmRemote(t, "evict-blob", srv.URL) + const path = "8/Everything/x86_64/Packages/a/a-1.0-1.el8.x86_64.rpm" + fetchBody(t, r, path) + if c := layers(t, r.Name, path); c != blobCached { + t.Fatalf("before evict = %+v, want %+v", c, blobCached) + } + if err := testEngine.Evict(context.Background(), r.Name, path); err != nil { + t.Fatalf("evict: %v", err) + } + if c := layers(t, r.Name, path); c != (cached{}) { + t.Fatalf("after evict = %+v, want every layer gone", c) + } +} + +func TestEvictRejectsNonDirectoryWildcard(t *testing.T) { requireStack(t) srv, _, _ := changingUpstream(t) r := rpmRemote(t, "evict-bare", srv.URL) const path = "8/Everything/x86_64/repodata/repomd.xml" fetchBody(t, r, path) - var pe *ProxyError - if err := testEngine.Evict(context.Background(), r.Name, "*"); !errors.As(err, &pe) || pe.Status != http.StatusBadRequest { - t.Fatalf("evict * = %v, want 400", err) + for _, bad := range []string{"*", "8*", "8/Every*"} { + var pe *ProxyError + if err := testEngine.Evict(context.Background(), r.Name, bad); !errors.As(err, &pe) || pe.Status != http.StatusBadRequest { + t.Errorf("evict %s = %v, want 400", bad, err) + } } - if fresh, _ := testEngine.cache.CheckTTL(context.Background(), r.Name, path); !fresh { - t.Error("bare * evicted cached keys") + if c := layers(t, r.Name, path); c != (indexCached) { + t.Errorf("rejected wildcard evicted layers: %+v", c) } } @@ -100,15 +152,22 @@ func TestEvictUnknownRemote(t *testing.T) { } } +func holdLock(t *testing.T, remote, path string) { + t.Helper() + ctx := context.Background() + if ok, err := testCache.AcquireLock(ctx, remote, path, time.Minute); !ok || err != nil { + t.Fatalf("acquire: %v %v", ok, err) + } + t.Cleanup(func() { _ = testCache.ReleaseLock(ctx, remote, path) }) +} + func TestEvictWaitsForFetchLock(t *testing.T) { requireStack(t) srv, _, _ := changingUpstream(t) r := rpmRemote(t, "evict-lock", srv.URL) ctx := context.Background() const path = "8/Everything/x86_64/repodata/repomd.xml" - if ok, err := testEngine.cache.AcquireLock(ctx, r.Name, path, time.Minute); !ok || err != nil { - t.Fatalf("acquire: %v %v", ok, err) - } + holdLock(t, r.Name, path) done := make(chan error, 1) go func() { done <- testEngine.Evict(ctx, r.Name, path) }() select { @@ -116,10 +175,60 @@ func TestEvictWaitsForFetchLock(t *testing.T) { t.Fatalf("evict returned while fetch lock held: %v", err) case <-time.After(200 * time.Millisecond): } - _ = testEngine.cache.ReleaseLock(ctx, r.Name, path) + _ = testCache.ReleaseLock(ctx, r.Name, path) if err := <-done; err != nil { t.Fatalf("evict: %v", err) } + ok, err := testCache.AcquireLock(ctx, r.Name, path, time.Second) + if !ok || err != nil { + t.Fatalf("lock still held after evict: %v %v", ok, err) + } +} + +func TestEvictCancelledWhileWaitingDeletesNothing(t *testing.T) { + requireStack(t) + srv, _, _ := changingUpstream(t) + r := rpmRemote(t, "evict-cancel", srv.URL) + const path = "8/Everything/x86_64/repodata/repomd.xml" + fetchBody(t, r, path) + holdLock(t, r.Name, path) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- testEngine.Evict(ctx, r.Name, path) }() + time.Sleep(100 * time.Millisecond) + cancel() + select { + case err := <-done: + if err == nil { + t.Fatal("cancelled evict returned nil") + } + case <-time.After(2 * time.Second): + t.Fatal("cancelled evict did not return") + } + if c := layers(t, r.Name, path); c != (indexCached) { + t.Errorf("cancelled evict deleted layers: %+v", c) + } +} + +func TestEvictLockTimeoutIs503(t *testing.T) { + requireStack(t) + srv, _, _ := changingUpstream(t) + r := rpmRemote(t, "evict-timeout", srv.URL) + const path = "8/Everything/x86_64/repodata/repomd.xml" + fetchBody(t, r, path) + holdLock(t, r.Name, path) + prev := testEngine.evictLockWait + testEngine.evictLockWait = 100 * time.Millisecond + t.Cleanup(func() { testEngine.evictLockWait = prev }) + + var pe *ProxyError + if err := testEngine.Evict(context.Background(), r.Name, path); !errors.As(err, &pe) || pe.Status != http.StatusServiceUnavailable { + t.Fatalf("evict = %v, want 503", err) + } + if c := layers(t, r.Name, path); c != (indexCached) { + t.Errorf("timed-out evict deleted layers: %+v", c) + } } func TestEvictWildcardClearsPrefixOnly(t *testing.T) { @@ -134,11 +243,22 @@ func TestEvictWildcardClearsPrefixOnly(t *testing.T) { for _, p := range []string{repomd, rpm, other} { fetchBody(t, r, p) } + if c := layers(t, r.Name, rpm); c != blobCached { + t.Fatalf("%s before evict = %+v, want %+v", rpm, c, blobCached) + } rev.Store("rev2") if err := testEngine.Evict(context.Background(), r.Name, "8/Everything/x86_64/*"); err != nil { t.Fatalf("evict: %v", err) } + for _, p := range []string{repomd, rpm} { + if c := layers(t, r.Name, p); c != (cached{}) { + t.Errorf("%s after evict = %+v, want every layer gone", p, c) + } + } + if c := layers(t, r.Name, other); c != (indexCached) { + t.Errorf("%s outside prefix = %+v, want %+v", other, c, indexCached) + } if got := fetchBody(t, r, repomd); got != "rev2:/"+repomd { t.Errorf("index under prefix = %q, want rev2", got) } diff --git a/internal/storage/s3.go b/internal/storage/s3.go index 1aa2ecb..817db94 100644 --- a/internal/storage/s3.go +++ b/internal/storage/s3.go @@ -99,15 +99,28 @@ func (s *S3) Stat(ctx context.Context, key string) (*minio.ObjectInfo, error) { return &info, nil } -// DeletePrefix removes every object whose key starts with prefix. +// DeletePrefix removes every object whose key starts with prefix. It lists +// first so a failed list returns its error instead of deleting nothing. func (s *S3) DeletePrefix(ctx context.Context, prefix string) error { - objects := s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: true}) - for res := range s.client.RemoveObjects(ctx, s.bucket, objects, minio.RemoveObjectsOptions{}) { - if res.Err != nil { - return res.Err + var objs []minio.ObjectInfo + for obj := range s.client.ListObjects(ctx, s.bucket, minio.ListObjectsOptions{Prefix: prefix, Recursive: true}) { + if obj.Err != nil { + return obj.Err + } + objs = append(objs, obj) + } + ch := make(chan minio.ObjectInfo, len(objs)) + for _, obj := range objs { + ch <- obj + } + close(ch) + var err error + for res := range s.client.RemoveObjects(ctx, s.bucket, ch, minio.RemoveObjectsOptions{}) { + if res.Err != nil && err == nil { + err = res.Err } } - return nil + return err } // ListStaleObjects returns keys under prefix last modified before cutoff. Used diff --git a/internal/storage/storage_test.go b/internal/storage/storage_test.go index 0379900..a5987e0 100644 --- a/internal/storage/storage_test.go +++ b/internal/storage/storage_test.go @@ -4,11 +4,15 @@ import ( "bytes" "context" "io" + "net/http" + "net/http/httptest" "os" "strings" "testing" "time" + "github.com/minio/minio-go/v7" + "git.unkin.net/unkin/artifactapi/internal/testsupport" ) @@ -158,3 +162,26 @@ func TestCASStore(t *testing.T) { t.Errorf("stored content mismatch: %q", got) } } + +// The fake endpoint fails every list but accepts every delete, as an S3 that +// tolerates deleting an empty key would. +func TestDeletePrefixReturnsListError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPost { + _, _ = io.WriteString(w, ``) + return + } + w.WriteHeader(http.StatusInternalServerError) + _, _ = io.WriteString(w, `InternalErrorlist failed`) + })) + defer srv.Close() + client, err := minio.New(strings.TrimPrefix(srv.URL, "http://"), &minio.Options{Region: "us-east-1", MaxRetries: 1}) + if err != nil { + t.Fatal(err) + } + s := &S3{client: client, bucket: "bucket"} + err = s.DeletePrefix(context.Background(), "indexes/r/") + if err == nil { + t.Fatal("DeletePrefix on a failing list returned nil") + } +}