Release Redis locks only while the caller still owns them (#156)

AcquireLock stored a constant value and ReleaseLock was an unconditional DEL, so a holder that outlived its TTL deleted the next holder's lock. This hits Fetch, Evict and gems push/yank.

- store a random owner token with SET NX PX and return it from AcquireLock
- release via a Lua compare-and-delete keyed on that token
- carry the token through Fetch, Evict and Engine.Lock's release func
- release the Fetch lock with an uncancelled context
- make Engine.Lock's busy 503 message generic

Reviewed-on: #156
Co-authored-by: unkin-agent <unkin-agent@unkin.net>
Co-committed-by: unkin-agent <unkin-agent@unkin.net>

internal/proxy/engine.go

@@ -134,10 +134,11 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p
 		slog.Warn("cache hit but S3 miss, re-fetching", "remote", remote.Name, "path", path)
 	}
 
-	locked, err := e.cache.AcquireLock(ctx, remote.Name, path, fetchLockTTL)
+	token, err := e.cache.AcquireLock(ctx, remote.Name, path, fetchLockTTL)
 	if err != nil {
 		slog.Warn("lock acquire failed", "error", err)
 	}
+	locked := token != ""
 
 	if !locked {
 		// Another request holds the fetch lock. Poll the store until the leader
@@ -151,7 +152,7 @@ func (e *Engine) Fetch(ctx context.Context, remote models.Remote, path string, p
 	}
 
 	if locked {
-		defer e.cache.ReleaseLock(ctx, remote.Name, path)
+		defer func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remote.Name, path, token) }()
 	}
 
 	if class == ClassMutable && remote.CheckMutable {
@@ -257,31 +258,32 @@ func (e *Engine) Evict(ctx context.Context, remoteName, path string) error {
 // release. It fails with a 503 when the lock cannot be taken within wait or
 // Redis errors.
 func (e *Engine) Lock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) (func(), error) {
-	if err := e.waitForLock(ctx, remoteName, path, wait, ttl); err != nil {
+	token, err := e.waitForLock(ctx, remoteName, path, wait, ttl)
+	if err != nil {
 		return nil, err
 	}
-	return func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path) }, nil
+	return func() { _ = e.cache.ReleaseLock(context.WithoutCancel(ctx), remoteName, path, token) }, nil
 }
 
-func (e *Engine) waitForLock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) error {
+func (e *Engine) waitForLock(ctx context.Context, remoteName, path string, wait, ttl time.Duration) (string, error) {
 	deadline := time.Now().Add(wait)
 	for {
-		ok, err := e.cache.AcquireLock(ctx, remoteName, path, ttl)
-		if ok {
-			return nil
+		token, err := e.cache.AcquireLock(ctx, remoteName, path, ttl)
+		if token != "" {
+			return token, nil
 		}
 		if ctx.Err() != nil {
-			return ctx.Err()
+			return "", ctx.Err()
 		}
 		if err != nil {
-			return &ProxyError{Status: http.StatusServiceUnavailable, Message: fmt.Sprintf("fetch lock: %v", err)}
+			return "", &ProxyError{Status: http.StatusServiceUnavailable, Message: fmt.Sprintf("lock: %v", err)}
 		}
 		if time.Now().After(deadline) {
-			return &ProxyError{Status: http.StatusServiceUnavailable, Message: "fetch in progress, retry evict"}
+			return "", &ProxyError{Status: http.StatusServiceUnavailable, Message: "resource busy, retry later"}
 		}
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
+			return "", ctx.Err()
 		case <-time.After(50 * time.Millisecond):
 		}
 	}

internal/proxy/evict_test.go

@@ -152,13 +152,15 @@ func TestEvictUnknownRemote(t *testing.T) {
 	}
 }
 
-func holdLock(t *testing.T, remote, path string) {
+func holdLock(t *testing.T, remote, path string) 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)
+	token, err := testCache.AcquireLock(ctx, remote, path, time.Minute)
+	if token == "" || err != nil {
+		t.Fatalf("acquire: %q %v", token, err)
 	}
-	t.Cleanup(func() { _ = testCache.ReleaseLock(ctx, remote, path) })
+	t.Cleanup(func() { _ = testCache.ReleaseLock(ctx, remote, path, token) })
+	return token
 }
 
 func TestEvictWaitsForFetchLock(t *testing.T) {
@@ -167,7 +169,7 @@ func TestEvictWaitsForFetchLock(t *testing.T) {
 	r := rpmRemote(t, "evict-lock", srv.URL)
 	ctx := context.Background()
 	const path = "8/Everything/x86_64/repodata/repomd.xml"
-	holdLock(t, r.Name, path)
+	token := holdLock(t, r.Name, path)
 	done := make(chan error, 1)
 	go func() { done <- testEngine.Evict(ctx, r.Name, path) }()
 	select {
@@ -175,13 +177,13 @@ func TestEvictWaitsForFetchLock(t *testing.T) {
 		t.Fatalf("evict returned while fetch lock held: %v", err)
 	case <-time.After(200 * time.Millisecond):
 	}
-	_ = testCache.ReleaseLock(ctx, r.Name, path)
+	_ = testCache.ReleaseLock(ctx, r.Name, path, token)
 	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)
+	token, err := testCache.AcquireLock(ctx, r.Name, path, time.Second)
+	if token == "" || err != nil {
+		t.Fatalf("lock still held after evict: %q %v", token, err)
 	}
 }