diff --git a/internal/api/v1/gems_handler_test.go b/internal/api/v1/gems_handler_test.go index 110b735..c1cc0cc 100644 --- a/internal/api/v1/gems_handler_test.go +++ b/internal/api/v1/gems_handler_test.go @@ -231,8 +231,9 @@ func TestGemsLockTimeoutIs503(t *testing.T) { if err != nil { t.Fatal(err) } - if ok, err := redis.AcquireLock(context.Background(), repo, "_gemlock/busy-1.0.0", time.Minute); !ok || err != nil { - t.Fatalf("hold lock = %v %v", ok, err) + token, err := redis.AcquireLock(context.Background(), repo, "_gemlock/busy-1.0.0", time.Minute) + if token == "" || err != nil { + t.Fatalf("hold lock = %q %v", token, err) } prev := gemsLockWait gemsLockWait = 100 * time.Millisecond @@ -244,7 +245,7 @@ func TestGemsLockTimeoutIs503(t *testing.T) { if code, _ := npmDo(t, http.MethodDelete, base+"api/v1/gems/yank?gem_name=busy&version=1.0.0", ""); code != http.StatusServiceUnavailable { t.Fatalf("yank under held lock = %d, want 503", code) } - if err := redis.ReleaseLock(context.Background(), repo, "_gemlock/busy-1.0.0"); err != nil { + if err := redis.ReleaseLock(context.Background(), repo, "_gemlock/busy-1.0.0", token); err != nil { t.Fatal(err) } if code, body := npmDo(t, http.MethodPost, base+"api/v1/gems", string(testGemFile(t, "busy", "1.0.0", "ruby"))); code != http.StatusOK { diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 8bcb380..a1e1dd3 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -61,18 +61,54 @@ func TestTTL(t *testing.T) { func TestLock(t *testing.T) { requireRedis(t) ctx := context.Background() - ok, err := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute) - if err != nil || !ok { - t.Fatalf("first acquire should succeed: %v %v", ok, err) + token, err := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute) + if err != nil || token == "" { + t.Fatalf("first acquire should succeed: %q %v", token, err) } - if ok, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); ok { + if again, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); again != "" { t.Error("second acquire should fail while held") } - if err := testRedis.ReleaseLock(ctx, "r", "lockpath"); err != nil { + if err := testRedis.ReleaseLock(ctx, "r", "lockpath", "not-the-owner"); err != nil { t.Fatal(err) } - if ok, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); !ok { - t.Error("acquire should succeed after release") + if again, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute); again != "" { + t.Error("release with a foreign token must not drop the lock") + } + if err := testRedis.ReleaseLock(ctx, "r", "lockpath", token); err != nil { + t.Fatal(err) + } + next, _ := testRedis.AcquireLock(ctx, "r", "lockpath", time.Minute) + if next == "" { + t.Fatal("acquire should succeed after release") + } + if next == token { + t.Error("tokens must differ between holders") + } +} + +func TestStaleReleaseKeepsNextHolder(t *testing.T) { + requireRedis(t) + ctx := context.Background() + stale, err := testRedis.AcquireLock(ctx, "r", "stale", 50*time.Millisecond) + if err != nil || stale == "" { + t.Fatalf("acquire: %q %v", stale, err) + } + time.Sleep(100 * time.Millisecond) + fresh, err := testRedis.AcquireLock(ctx, "r", "stale", time.Minute) + if err != nil || fresh == "" { + t.Fatalf("acquire after expiry: %q %v", fresh, err) + } + if err := testRedis.ReleaseLock(ctx, "r", "stale", stale); err != nil { + t.Fatal(err) + } + if again, _ := testRedis.AcquireLock(ctx, "r", "stale", time.Minute); again != "" { + t.Fatal("stale holder's release dropped the second holder's lock") + } + if err := testRedis.ReleaseLock(ctx, "r", "stale", fresh); err != nil { + t.Fatal(err) + } + if again, _ := testRedis.AcquireLock(ctx, "r", "stale", time.Minute); again == "" { + t.Error("owner release should free the lock") } } diff --git a/internal/cache/redis.go b/internal/cache/redis.go index eeca6c5..c120745 100644 --- a/internal/cache/redis.go +++ b/internal/cache/redis.go @@ -2,6 +2,7 @@ package cache import ( "context" + "crypto/rand" "fmt" "strings" "time" @@ -46,15 +47,23 @@ func (r *Redis) CheckTTL(ctx context.Context, remote, path string) (bool, error) return exists > 0, nil } -func (r *Redis) AcquireLock(ctx context.Context, remote, path string, ttl time.Duration) (bool, error) { +// AcquireLock returns the owner token on success and "" when the lock is held. +func (r *Redis) AcquireLock(ctx context.Context, remote, path string, ttl time.Duration) (string, error) { key := fmt.Sprintf("lock:%s:%s", remote, path) - ok, err := r.client.SetNX(ctx, key, "1", ttl).Result() - return ok, err + token := rand.Text() + ok, err := r.client.SetNX(ctx, key, token, ttl).Result() + if err != nil || !ok { + return "", err + } + return token, nil } -func (r *Redis) ReleaseLock(ctx context.Context, remote, path string) error { +var releaseScript = redis.NewScript(`if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end`) + +// ReleaseLock deletes the lock only while token still owns it. +func (r *Redis) ReleaseLock(ctx context.Context, remote, path, token string) error { key := fmt.Sprintf("lock:%s:%s", remote, path) - return r.client.Del(ctx, key).Err() + return releaseScript.Run(ctx, r.client, []string{key}, token).Err() } // SetETag stores etag for path, or drops the stored one when etag is empty. diff --git a/internal/proxy/engine.go b/internal/proxy/engine.go index d097733..99d24b7 100644 --- a/internal/proxy/engine.go +++ b/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): } } diff --git a/internal/proxy/evict_test.go b/internal/proxy/evict_test.go index 622cc35..55635f4 100644 --- a/internal/proxy/evict_test.go +++ b/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) } }