Restrict wildcard evict to <dir>/*, return 503 on lock contention, surface S3 list errors

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 <dir>/*"}, 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)
+		}
 	}
 }

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 <dir>/*"}
 	}
 	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):
 		}
 	}

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)
 	}

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

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, `<DeleteResult></DeleteResult>`)
+			return
+		}
+		w.WriteHeader(http.StatusInternalServerError)
+		_, _ = io.WriteString(w, `<Error><Code>InternalError</Code><Message>list failed</Message></Error>`)
+	}))
+	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")
+	}
+}