diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index 99ee449afd..b9fdf46dd1 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -872,6 +872,7 @@ message Lock { int64 generation = 5; bool is_backup = 6; int64 seq = 7; + string prev_renew_token = 8; } message TransferLocksRequest { repeated Lock locks = 1; @@ -886,6 +887,7 @@ message ReplicateLockRequest { int64 generation = 5; bool is_unlock = 6; int64 seq = 7; + string prev_renew_token = 8; } message ReplicateLockResponse { } diff --git a/weed/cluster/lock_client.go b/weed/cluster/lock_client.go index 9d14f383f2..2574de14e9 100644 --- a/weed/cluster/lock_client.go +++ b/weed/cluster/lock_client.go @@ -45,6 +45,12 @@ type LockClient struct { // snapshot) but short enough that a write fails instead of hanging // forever. noLockServerRetryPeriod time.Duration + + // lockRpcTimeout bounds a single DistributedLock attempt: a ring member + // reachable from the masters but not from this filer would otherwise + // block an attempt on the TCP connect timeout, far past + // noLockServerRetryPeriod. + lockRpcTimeout time.Duration } func NewLockClient(grpcDialOption grpc.DialOption, seedFiler pb.ServerAddress) *LockClient { @@ -55,6 +61,7 @@ func NewLockClient(grpcDialOption grpc.DialOption, seedFiler pb.ServerAddress) * seedFiler: seedFiler, priorWindow: 5 * time.Second, noLockServerRetryPeriod: 15 * time.Second, + lockRpcTimeout: 10 * time.Second, } } @@ -304,39 +311,78 @@ func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerCh // waiter must pick it up promptly, otherwise cross-mount write handoff stalls // long enough to time out clients. func (lock *LiveLock) retryUntilLocked(lockDuration time.Duration) error { - var unavailableSince time.Time - for lock.renewToken == "" { - err := lock.AttemptToLock(lockDuration) + // The retry window bounds the attempts against one route: the clock + // starts before the first attempt and each attempt gets what is left + // of it. A spent window on a non-seed target falls back to the seed + // filer — which can forward to the real owner — and that route gets + // its own window, so a reachable fallback is always tried. + unavailableSince := time.Now() + // The seed route gets one fresh window per acquisition: the seed can + // redirect back to a dead primary via LockHostMovedTo, and refilling the + // window on every return to the seed would make that cycle unbounded. + seedRetried := false + var err error + for { + attemptTimeout := lock.lc.lockRpcTimeout + if remaining := lock.lc.noLockServerRetryPeriod - time.Since(unavailableSince); remaining < attemptTimeout { + if remaining <= 0 { + if lock.hostFiler.Equals(lock.lc.seedFiler) || seedRetried { + if err == nil { + err = fmt.Errorf("lock %s: no lock server granted within %s", lock.key, lock.lc.noLockServerRetryPeriod) + } + return err + } + glog.V(0).Infof("LOCK: no response within %s for key=%s filer=%s, falling back to seed filer=%s", + lock.lc.noLockServerRetryPeriod, lock.key, lock.hostFiler, lock.lc.seedFiler) + lock.hostFiler = lock.lc.seedFiler + seedRetried = true + unavailableSince = time.Now() + continue + } + attemptTimeout = remaining + } + hostBefore := lock.hostFiler + var granted bool + granted, err = lock.attemptToLockWithin(lockDuration, attemptTimeout) if err == nil { - unavailableSince = time.Time{} + if granted { + return nil + } + // A clean LockHostMovedTo redirect returns no error and no + // verdict; it must still count against the window, or a redirect + // cycle between ring-disagreeing filers retries forever. continue } glog.V(1).Infof("create lock %s: %v", lock.key, err) if strings.Contains(err.Error(), "lock already owned") { // Ordinary contention: a reachable server holds the lock, so // waiting is the point and has no bound. - unavailableSince = time.Time{} + unavailableSince = time.Now() continue } // Anything else — "no lock server found", a dead ring member refusing - // connections — is a systemic fault waiting cannot fix; give up once - // it persists past the retry period. - if unavailableSince.IsZero() { + // connections — is a systemic fault waiting cannot fix. A failed + // attempt that fell back to the seed route gets a fresh window for + // it; a server redirect stays inside the current one. + if !seedRetried && lock.hostFiler.Equals(lock.lc.seedFiler) && !hostBefore.Equals(lock.lc.seedFiler) { + seedRetried = true unavailableSince = time.Now() - } else if time.Since(unavailableSince) > lock.lc.noLockServerRetryPeriod { - return err } } - return nil } func (lock *LiveLock) AttemptToLock(lockDuration time.Duration) error { + _, err := lock.attemptToLockWithin(lockDuration, lock.lc.lockRpcTimeout) + return err +} + +func (lock *LiveLock) attemptToLockWithin(lockDuration, attemptTimeout time.Duration) (granted bool, err error) { glog.V(4).Infof("LOCK: AttemptToLock key=%s owner=%s", lock.key, lock.self) - errorMessage, err := lock.doLock(lockDuration) + errorMessage, granted, err := lock.doLock(lockDuration, attemptTimeout) if err != nil { glog.V(1).Infof("LOCK: doLock failed for key=%s: %v", lock.key, err) time.Sleep(time.Second) - return err + return false, err } if errorMessage != "" { if strings.Contains(errorMessage, "lock already owned") { @@ -345,14 +391,21 @@ func (lock *LiveLock) AttemptToLock(lockDuration time.Duration) error { glog.V(2).Infof("LOCK: doLock returned error message for key=%s: %s", lock.key, errorMessage) } time.Sleep(time.Second) - return fmt.Errorf("%v", errorMessage) + return false, fmt.Errorf("%v", errorMessage) + } + if !granted { + // A bare LockHostMovedTo redirect carries no verdict: nothing was + // acquired, so leave isLocked as it is and keep the same ~1s cadence + // as other retries. + time.Sleep(time.Second) + return false, nil } if atomic.LoadInt32(&lock.isLocked) == 0 { // Only log when transitioning from unlocked to locked glog.V(1).Infof("LOCK: Successfully acquired key=%s owner=%s", lock.key, lock.self) } atomic.StoreInt32(&lock.isLocked, 1) - return nil + return true, nil } func (lock *LiveLock) StopShortLivedLock() error { @@ -363,7 +416,9 @@ func (lock *LiveLock) StopShortLivedLock() error { atomic.StoreInt32(&lock.isLocked, 0) }() return pb.WithFilerClient(false, 0, lock.hostFiler, lock.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - _, err := client.DistributedUnlock(context.Background(), &filer_pb.UnlockRequest{ + ctx, cancel := context.WithTimeout(context.Background(), lock.lc.lockRpcTimeout) + defer cancel() + _, err := client.DistributedUnlock(ctx, &filer_pb.UnlockRequest{ Name: lock.key, RenewToken: lock.renewToken, }) @@ -409,7 +464,7 @@ func (lock *LiveLock) Stop() error { return lock.StopShortLivedLock() } -func (lock *LiveLock) doLock(lockDuration time.Duration) (errorMessage string, err error) { +func (lock *LiveLock) doLock(lockDuration, attemptTimeout time.Duration) (errorMessage string, granted bool, err error) { glog.V(4).Infof("LOCK: doLock calling DistributedLock - key=%s filer=%s owner=%s", lock.key, lock.hostFiler, lock.self) @@ -417,7 +472,11 @@ func (lock *LiveLock) doLock(lockDuration time.Duration) (errorMessage string, e previousOwner := lock.owner err = pb.WithFilerClient(false, 0, lock.hostFiler, lock.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - resp, err := client.DistributedLock(context.Background(), &filer_pb.LockRequest{ + // Bound each attempt: a ring member that is reachable from the masters + // but not from this filer would otherwise block on the connect timeout. + ctx, cancel := context.WithTimeout(context.Background(), attemptTimeout) + defer cancel() + resp, err := client.DistributedLock(ctx, &filer_pb.LockRequest{ Name: lock.key, SecondsToLock: int64(lockDuration.Seconds()), RenewToken: lock.renewToken, @@ -426,16 +485,25 @@ func (lock *LiveLock) doLock(lockDuration time.Duration) (errorMessage string, e }) glog.V(4).Infof("LOCK: DistributedLock response - key=%s err=%v", lock.key, err) if err == nil && resp != nil { - lock.renewToken = resp.RenewToken + // Only a verdict replaces the token: a grant carries the new + // token, a rejection carries an error. A bare LockHostMovedTo + // redirect answers neither and must not drop a held token — + // the next attempt can still renew with it on the new host. + granted = resp.Error == "" && resp.RenewToken != "" + if resp.RenewToken != "" || resp.Error != "" { + lock.renewToken = resp.RenewToken + } if resp.Generation > 0 { atomic.StoreInt64(&lock.generation, resp.Generation) } lock.consecutiveFailures = 0 // Reset failure counter on success glog.V(4).Infof("LOCK: Got renewToken for key=%s", lock.key) } else { - //this can be retried. Need to remember the last valid renewToken - lock.renewToken = "" - glog.V(1).Infof("LOCK: Cleared renewToken for key=%s (err=%v)", lock.key, err) + // The request may have renewed the lease server-side before its + // response was lost: the last token can still be valid, so the + // next attempt retries with it. A definitive rejection clears it + // through the response's empty RenewToken. + glog.V(1).Infof("LOCK: no response for key=%s (err=%v)", lock.key, err) } if resp != nil { errorMessage = resp.Error @@ -469,7 +537,6 @@ func (lock *LiveLock) doLock(lockDuration time.Duration) (errorMessage string, e lock.consecutiveFailures, lock.key, lock.hostFiler, lock.lc.seedFiler) lock.hostFiler = lock.lc.seedFiler lock.consecutiveFailures = 0 - lock.renewToken = "" } } diff --git a/weed/cluster/lock_client_test.go b/weed/cluster/lock_client_test.go index 7a6a89950f..796cf28021 100644 --- a/weed/cluster/lock_client_test.go +++ b/weed/cluster/lock_client_test.go @@ -286,6 +286,307 @@ func TestNewShortLivedLockFailsFastOnUnreachableFiler(t *testing.T) { } } +type hangingLockFiler struct { + filer_pb.UnimplementedSeaweedFilerServer +} + +func (s *hangingLockFiler) DistributedLock(ctx context.Context, req *filer_pb.LockRequest) (*filer_pb.LockResponse, error) { + <-ctx.Done() + return nil, ctx.Err() +} + +type grantingLockFiler struct { + filer_pb.UnimplementedSeaweedFilerServer +} + +func (s *grantingLockFiler) DistributedLock(ctx context.Context, req *filer_pb.LockRequest) (*filer_pb.LockResponse, error) { + return &filer_pb.LockResponse{RenewToken: "tok"}, nil +} + +// A ring member that accepts connections but never answers must not swallow +// the retry window: when its budget is spent the seed route still gets an +// attempt before acquisition gives up. +func TestNewShortLivedLockFallsBackToSeedOnHangingPrimary(t *testing.T) { + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + + hangListener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + hangServer := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(hangServer, &hangingLockFiler{}) + go hangServer.Serve(hangListener) + defer hangServer.Stop() + + seedListener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + seedServer := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(seedServer, &grantingLockFiler{}) + go seedServer.Serve(seedListener) + defer seedServer.Stop() + + hangHost, hangPort, err := net.SplitHostPort(hangListener.Addr().String()) + if err != nil { + t.Fatalf("split host port: %v", err) + } + seedHost, seedPort, err := net.SplitHostPort(seedListener.Addr().String()) + if err != nil { + t.Fatalf("split host port: %v", err) + } + // ServerAddress.Equals compares the http port, so the two fake filers + // need distinct ones even though only the grpc port is dialed. + seedAddr := pb.ServerAddress(fmt.Sprintf("%s:1.%s", seedHost, seedPort)) + primaryAddr := pb.ServerAddress(fmt.Sprintf("%s:2.%s", hangHost, hangPort)) + + lc := NewLockClient(dialOption, seedAddr) + lc.lockRpcTimeout = 50 * time.Millisecond + lc.noLockServerRetryPeriod = 200 * time.Millisecond + lc.SetRing([]pb.ServerAddress{primaryAddr}, 1) + + start := time.Now() + lock := lc.NewShortLivedLock("test-key", "test-owner") + elapsed := time.Since(start) + + if lock == nil { + t.Fatal("expected the seed fallback to acquire the lock") + } + if !lock.hostFiler.Equals(seedAddr) { + t.Fatalf("expected acquisition through the seed %s, got %s", seedAddr, lock.hostFiler) + } + if elapsed > 5*time.Second { + t.Fatalf("lock acquisition took %v, expected it bounded by the per-route window", elapsed) + } +} + +type redirectingLockFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + movedTo string +} + +func (s *redirectingLockFiler) DistributedLock(ctx context.Context, req *filer_pb.LockRequest) (*filer_pb.LockResponse, error) { + return &filer_pb.LockResponse{Error: "forward failed", LockHostMovedTo: s.movedTo}, nil +} + +// The seed can redirect back to a dead primary via LockHostMovedTo. The seed +// fallback gets one fresh window per acquisition — letting every return to +// the seed refill it would make the primary->seed->primary cycle unbounded. +func TestNewShortLivedLockBoundsSeedRedirectCycle(t *testing.T) { + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + + hangListener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + hangServer := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(hangServer, &hangingLockFiler{}) + go hangServer.Serve(hangListener) + defer hangServer.Stop() + + hangHost, hangPort, err := net.SplitHostPort(hangListener.Addr().String()) + if err != nil { + t.Fatalf("split host port: %v", err) + } + primaryAddr := pb.ServerAddress(fmt.Sprintf("%s:2.%s", hangHost, hangPort)) + + seedListener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + seedServer := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(seedServer, &redirectingLockFiler{movedTo: string(primaryAddr)}) + go seedServer.Serve(seedListener) + defer seedServer.Stop() + + seedHost, seedPort, err := net.SplitHostPort(seedListener.Addr().String()) + if err != nil { + t.Fatalf("split host port: %v", err) + } + seedAddr := pb.ServerAddress(fmt.Sprintf("%s:1.%s", seedHost, seedPort)) + + lc := NewLockClient(dialOption, seedAddr) + lc.lockRpcTimeout = 50 * time.Millisecond + lc.noLockServerRetryPeriod = 200 * time.Millisecond + + start := time.Now() + lock := lc.NewShortLivedLock("test-key", "test-owner") + elapsed := time.Since(start) + + if lock != nil { + t.Fatal("expected nil lock when the seed keeps redirecting to a dead primary") + } + if elapsed > 5*time.Second { + t.Fatalf("lock acquisition took %v, expected a bounded redirect cycle", elapsed) + } +} + +type cleanRedirectLockFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + movedTo string +} + +func (s *cleanRedirectLockFiler) DistributedLock(ctx context.Context, req *filer_pb.LockRequest) (*filer_pb.LockResponse, error) { + return &filer_pb.LockResponse{LockHostMovedTo: s.movedTo}, nil +} + +// A second hop whose ring disagrees answers with a bare LockHostMovedTo and +// no error, so the first hop relays a clean redirect. Two filers pointing at +// each other then bounce the client forever if every redirect refills the +// window — tokenless redirects must count against it like failures do. +func TestNewShortLivedLockBoundsCleanRedirectCycle(t *testing.T) { + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + + listenerA, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + listenerB, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + hostA, portA, err := net.SplitHostPort(listenerA.Addr().String()) + if err != nil { + t.Fatalf("split host port: %v", err) + } + hostB, portB, err := net.SplitHostPort(listenerB.Addr().String()) + if err != nil { + t.Fatalf("split host port: %v", err) + } + addrA := pb.ServerAddress(fmt.Sprintf("%s:1.%s", hostA, portA)) + addrB := pb.ServerAddress(fmt.Sprintf("%s:2.%s", hostB, portB)) + + serverA := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(serverA, &cleanRedirectLockFiler{movedTo: string(addrB)}) + go serverA.Serve(listenerA) + defer serverA.Stop() + serverB := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(serverB, &cleanRedirectLockFiler{movedTo: string(addrA)}) + go serverB.Serve(listenerB) + defer serverB.Stop() + + lc := NewLockClient(dialOption, addrA) + lc.lockRpcTimeout = 50 * time.Millisecond + lc.noLockServerRetryPeriod = 200 * time.Millisecond + + done := make(chan *LiveLock, 1) + go func() { + done <- lc.NewShortLivedLock("test-key", "test-owner") + }() + select { + case lock := <-done: + if lock != nil { + t.Fatal("expected nil lock when the ring keeps redirecting") + } + case <-time.After(5 * time.Second): + t.Fatal("clean redirect cycle was not bounded by the retry window") + } +} + +// A failed attempt keeps the token it sent: the request may have renewed the +// lease server-side before its response was lost, and a retry with the last +// token can still succeed. Only a definitive rejection may clear it. +func TestDoLockKeepsRenewTokenOnError(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + addr := listener.Addr().String() + listener.Close() + + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + host, port, err := net.SplitHostPort(addr) + if err != nil { + t.Fatalf("split host port: %v", err) + } + filerAddr := pb.ServerAddress(fmt.Sprintf("%s:0.%s", host, port)) + lc := NewLockClient(dialOption, filerAddr) + lock := &LiveLock{ + key: "test-key", + renewToken: "tok", + hostFiler: filerAddr, + grpcDialOption: dialOption, + lc: lc, + } + if _, _, err := lock.doLock(time.Second, 50*time.Millisecond); err == nil { + t.Fatal("expected the attempt to fail against a dead member") + } + if lock.renewToken != "tok" { + t.Errorf("renewToken = %q, want it kept for the next attempt", lock.renewToken) + } +} + +// A definitive rejection answers with an empty token, which clears the stale +// one — only transport failures keep it. +func TestDoLockClearsRenewTokenOnRejection(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + grpcServer := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(grpcServer, &contendedLockFiler{}) + go grpcServer.Serve(listener) + defer grpcServer.Stop() + + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + host, port, err := net.SplitHostPort(listener.Addr().String()) + if err != nil { + t.Fatalf("split host port: %v", err) + } + filerAddr := pb.ServerAddress(fmt.Sprintf("%s:0.%s", host, port)) + lc := NewLockClient(dialOption, filerAddr) + lock := &LiveLock{ + key: "test-key", + renewToken: "tok", + hostFiler: filerAddr, + grpcDialOption: dialOption, + lc: lc, + } + errorMessage, _, err := lock.doLock(time.Second, time.Second) + if err != nil || errorMessage == "" { + t.Fatalf("expected a contended attempt to report an error message, got errorMessage=%q err=%v", errorMessage, err) + } + if lock.renewToken != "" { + t.Errorf("renewToken = %q, want it cleared by the rejection", lock.renewToken) + } +} + +// A bare LockHostMovedTo redirect is not a verdict on the token: the +// request may have committed a renewal before the ring moved, so the held +// token survives to renew against the new host. +func TestDoLockKeepsRenewTokenOnCleanRedirect(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + grpcServer := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(grpcServer, &cleanRedirectLockFiler{movedTo: "127.0.0.1:1.1"}) + go grpcServer.Serve(listener) + defer grpcServer.Stop() + + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + host, port, err := net.SplitHostPort(listener.Addr().String()) + if err != nil { + t.Fatalf("split host port: %v", err) + } + filerAddr := pb.ServerAddress(fmt.Sprintf("%s:0.%s", host, port)) + lc := NewLockClient(dialOption, filerAddr) + lock := &LiveLock{ + key: "test-key", + renewToken: "tok", + hostFiler: filerAddr, + grpcDialOption: dialOption, + lc: lc, + } + errorMessage, _, err := lock.doLock(time.Second, time.Second) + if err != nil || errorMessage != "" { + t.Fatalf("expected a clean redirect, got errorMessage=%q err=%v", errorMessage, err) + } + if lock.renewToken != "tok" { + t.Errorf("renewToken = %q, want it kept across the redirect", lock.renewToken) + } +} + // LiveLock.generation is a fencing token written and read with 64-bit atomic // operations. On 32-bit platforms (GOARCH=386 and GOARCH=arm) a 64-bit atomic // op requires an 8-byte-aligned address, which Go only guarantees for the diff --git a/weed/cluster/lock_manager/distributed_lock_manager.go b/weed/cluster/lock_manager/distributed_lock_manager.go index 6c50e98563..0bfb89971c 100644 --- a/weed/cluster/lock_manager/distributed_lock_manager.go +++ b/weed/cluster/lock_manager/distributed_lock_manager.go @@ -17,7 +17,7 @@ var NoLockServerError = fmt.Errorf("no lock server found") // The caller (filer server) provides this to avoid a circular dependency. // seq is a per-lock monotonic sequence number for causal ordering — the backup // rejects mutations with seq <= its current seq for that key. -type ReplicateFunc func(server pb.ServerAddress, key string, expiredAtNs int64, token string, owner string, generation int64, seq int64, isUnlock bool) +type ReplicateFunc func(server pb.ServerAddress, key string, expiredAtNs int64, token string, prevToken string, owner string, generation int64, seq int64, isUnlock bool) type DistributedLockManager struct { lockManager *LockManager @@ -46,11 +46,11 @@ func (dlm *DistributedLockManager) LockWithTimeout(key string, expiredAtNs int64 // ring update and lock transfer completion — the old primary remains // authoritative for locks it still holds. if token != "" { - if lock, found := dlm.lockManager.GetLock(key); found && !lock.IsBackup && lock.Token == token { + if lock, found := dlm.lockManager.GetLock(key); found && !lock.IsBackup && (lock.Token == token || lock.PrevToken == token) { var seq int64 lockOwner, renewToken, generation, seq, err = dlm.lockManager.Lock(key, expiredAtNs, token, owner) if err == nil && renewToken != "" { - dlm.replicateToBackup(key, expiredAtNs, renewToken, owner, generation, seq, false) + dlm.replicateToBackup(key, expiredAtNs, renewToken, token, owner, generation, seq, false) } return } @@ -61,7 +61,7 @@ func (dlm *DistributedLockManager) LockWithTimeout(key string, expiredAtNs int64 var seq int64 lockOwner, renewToken, generation, seq, err = dlm.lockManager.Lock(key, expiredAtNs, token, owner) if err == nil && renewToken != "" { - dlm.replicateToBackup(key, expiredAtNs, renewToken, owner, generation, seq, false) + dlm.replicateToBackup(key, expiredAtNs, renewToken, token, owner, generation, seq, false) } return } @@ -95,13 +95,13 @@ func (dlm *DistributedLockManager) Unlock(key string, token string) (movedTo pb. } if primary != dlm.Host { // If we still hold this lock locally, serve the unlock here - if lock, found := dlm.lockManager.GetLock(key); found && !lock.IsBackup && lock.Token == token { + if lock, found := dlm.lockManager.GetLock(key); found && !lock.IsBackup && token != "" && (lock.Token == token || lock.PrevToken == token) { var isUnlocked bool var generation int64 var seq int64 isUnlocked, generation, seq, err = dlm.lockManager.Unlock(key, token) if isUnlocked { - dlm.replicateToBackup(key, 0, "", "", generation, seq, true) + dlm.replicateToBackup(key, 0, "", "", "", generation, seq, true) } return } @@ -113,7 +113,7 @@ func (dlm *DistributedLockManager) Unlock(key string, token string) (movedTo pb. var seq int64 isUnlocked, generation, seq, err = dlm.lockManager.Unlock(key, token) if isUnlocked { - dlm.replicateToBackup(key, 0, "", "", generation, seq, true) + dlm.replicateToBackup(key, 0, "", "", "", generation, seq, true) } return } @@ -121,15 +121,15 @@ func (dlm *DistributedLockManager) Unlock(key string, token string) (movedTo pb. // InsertLock is used to insert a lock to a server unconditionally. // It is used when a server is down and the lock is moved to another server. // After inserting, it replicates to the backup for this key. -func (dlm *DistributedLockManager) InsertLock(key string, expiredAtNs int64, token string, owner string, generation int64, seq int64) { - if dlm.lockManager.InsertLock(key, expiredAtNs, token, owner, generation, seq) { - dlm.replicateToBackup(key, expiredAtNs, token, owner, generation, seq, false) +func (dlm *DistributedLockManager) InsertLock(key string, expiredAtNs int64, token string, prevToken string, owner string, generation int64, seq int64) { + if dlm.lockManager.InsertLock(key, expiredAtNs, token, prevToken, owner, generation, seq) { + dlm.replicateToBackup(key, expiredAtNs, token, prevToken, owner, generation, seq, false) } } // InsertBackupLock inserts a lock as a backup copy, rejecting stale seq -func (dlm *DistributedLockManager) InsertBackupLock(key string, expiredAtNs int64, token string, owner string, generation int64, seq int64) { - dlm.lockManager.InsertBackupLock(key, expiredAtNs, token, owner, generation, seq) +func (dlm *DistributedLockManager) InsertBackupLock(key string, expiredAtNs int64, token string, prevToken string, owner string, generation int64, seq int64) { + dlm.lockManager.InsertBackupLock(key, expiredAtNs, token, prevToken, owner, generation, seq) } // RemoveBackupLock removes a backup lock unconditionally @@ -181,12 +181,12 @@ func (dlm *DistributedLockManager) GetLock(key string) (*Lock, bool) { } // replicateToBackup asynchronously replicates a lock operation to the backup server -func (dlm *DistributedLockManager) replicateToBackup(key string, expiredAtNs int64, token string, owner string, generation int64, seq int64, isUnlock bool) { +func (dlm *DistributedLockManager) replicateToBackup(key string, expiredAtNs int64, token string, prevToken string, owner string, generation int64, seq int64, isUnlock bool) { _, backup := dlm.LockRing.GetPrimaryAndBackup(key) if backup == "" { return // single-server deployment, no backup } if dlm.ReplicateFn != nil { - go dlm.ReplicateFn(backup, key, expiredAtNs, token, owner, generation, seq, isUnlock) + go dlm.ReplicateFn(backup, key, expiredAtNs, token, prevToken, owner, generation, seq, isUnlock) } } diff --git a/weed/cluster/lock_manager/distributed_lock_manager_test.go b/weed/cluster/lock_manager/distributed_lock_manager_test.go index 652532fbae..7c03b6cbce 100644 --- a/weed/cluster/lock_manager/distributed_lock_manager_test.go +++ b/weed/cluster/lock_manager/distributed_lock_manager_test.go @@ -33,7 +33,7 @@ func newTestCluster(hosts ...pb.ServerAddress) *testCluster { // Wire up replication: each node's ReplicateFn calls the backup's DLM directly for _, dlm := range c.nodes { d := dlm // capture - d.ReplicateFn = func(server pb.ServerAddress, key string, expiredAtNs int64, token string, owner string, generation int64, seq int64, isUnlock bool) { + d.ReplicateFn = func(server pb.ServerAddress, key string, expiredAtNs int64, token string, prevToken string, owner string, generation int64, seq int64, isUnlock bool) { c.mu.Lock() target, ok := c.nodes[server] c.mu.Unlock() @@ -43,7 +43,7 @@ func newTestCluster(hosts ...pb.ServerAddress) *testCluster { if isUnlock { target.RemoveBackupLockIfSeq(key, generation, seq) } else { - target.InsertBackupLock(key, expiredAtNs, token, owner, generation, seq) + target.InsertBackupLock(key, expiredAtNs, token, prevToken, owner, generation, seq) } } } @@ -70,7 +70,7 @@ func (c *testCluster) addNode(host pb.ServerAddress) { c.mu.Unlock() // Wire up replication - dlm.ReplicateFn = func(server pb.ServerAddress, key string, expiredAtNs int64, token string, owner string, generation int64, seq int64, isUnlock bool) { + dlm.ReplicateFn = func(server pb.ServerAddress, key string, expiredAtNs int64, token string, prevToken string, owner string, generation int64, seq int64, isUnlock bool) { c.mu.Lock() target, ok := c.nodes[server] c.mu.Unlock() @@ -80,7 +80,7 @@ func (c *testCluster) addNode(host pb.ServerAddress) { if isUnlock { target.RemoveBackupLockIfSeq(key, generation, seq) } else { - target.InsertBackupLock(key, expiredAtNs, token, owner, generation, seq) + target.InsertBackupLock(key, expiredAtNs, token, prevToken, owner, generation, seq) } } @@ -311,7 +311,7 @@ func TestDLM_RollingRestart(t *testing.T) { for _, lock := range dlm.AllLocks() { newPrimary, _ := dlm.LockRing.GetPrimaryAndBackup(lock.Key) if newPrimary == dlm.Host && !lock.IsBackup { - dlm.replicateToBackup(lock.Key, lock.ExpiredAtNs, lock.Token, lock.Owner, lock.Generation, lock.Seq, false) + dlm.replicateToBackup(lock.Key, lock.ExpiredAtNs, lock.Token, lock.PrevToken, lock.Owner, lock.Generation, lock.Seq, false) } } } @@ -370,7 +370,7 @@ func TestDLM_ReplicationFailure_PrimaryStillWorks(t *testing.T) { // Break replication by setting a no-op ReplicateFn on all nodes for _, dlm := range cluster.getNodes() { - dlm.ReplicateFn = func(server pb.ServerAddress, key string, expiredAtNs int64, token string, owner string, generation int64, seq int64, isUnlock bool) { + dlm.ReplicateFn = func(server pb.ServerAddress, key string, expiredAtNs int64, token string, prevToken string, owner string, generation int64, seq int64, isUnlock bool) { // Simulate replication failure: do nothing } } @@ -539,7 +539,7 @@ func TestDLM_NodeDropAndJoin_OwnershipDisruption(t *testing.T) { for _, lock := range dlm.AllLocks() { p, _ := dlm.LockRing.GetPrimaryAndBackup(lock.Key) if p == dlm.Host && !lock.IsBackup { - dlm.replicateToBackup(lock.Key, lock.ExpiredAtNs, lock.Token, lock.Owner, lock.Generation, lock.Seq, false) + dlm.replicateToBackup(lock.Key, lock.ExpiredAtNs, lock.Token, lock.PrevToken, lock.Owner, lock.Generation, lock.Seq, false) } } } @@ -558,7 +558,7 @@ func TestDLM_NodeDropAndJoin_OwnershipDisruption(t *testing.T) { // This lock should move to the new primary target := cluster.get(p) if target != nil { - target.InsertLock(lock.Key, lock.ExpiredAtNs, lock.Token, lock.Owner, lock.Generation, lock.Seq) + target.InsertLock(lock.Key, lock.ExpiredAtNs, lock.Token, lock.PrevToken, lock.Owner, lock.Generation, lock.Seq) dlm.DemoteLock(lock.Key) } } @@ -671,26 +671,96 @@ func TestDLM_RenewalDuringTransferWindow(t *testing.T) { t.Logf("Key %s: primary changed from %s to %s, but renewal on old primary succeeded", key, primaryHost, newPrimary) } +func TestDLM_ReplayedRenewalExtendsBackupExpiry(t *testing.T) { + hosts := []pb.ServerAddress{"filer1:8888", "filer2:8888", "filer3:8888"} + cluster := newTestCluster(hosts...) + + key := "replay-expiry-lock" + token1, _, primaryHost, err := cluster.acquireLock(key, "owner1", 30*time.Second) + require.NoError(t, err) + + _, backup := cluster.get(primaryHost).LockRing.GetPrimaryAndBackup(key) + require.NotEmpty(t, backup) + backupDlm := cluster.get(backup) + primary := cluster.get(primaryHost) + + // Renewal rotates to token2, replicating seq=2 with expiry e1. + e1 := time.Now().Add(30 * time.Second).UnixNano() + _, token2, _, _, err := primary.LockWithTimeout(key, e1, token1, "owner1") + require.NoError(t, err) + time.Sleep(10 * time.Millisecond) + + // The renewal response was lost; the client retries with the superseded + // token and a later expiry e2. + e2 := time.Now().Add(60 * time.Second).UnixNano() + _, replayed, _, _, err := primary.LockWithTimeout(key, e2, token1, "owner1") + require.NoError(t, err) + assert.Equal(t, token2, replayed) + + // The backup must accept the extended expiry instead of rejecting the + // replayed mutation as stale. + require.Eventually(t, func() bool { + lock, found := backupDlm.GetLock(key) + return found && lock.ExpiredAtNs == e2 && lock.Seq == 3 + }, time.Second, time.Millisecond) +} + +func TestDLM_BackupPromotionKeepsLostRenewalRecovery(t *testing.T) { + hosts := []pb.ServerAddress{"filer1:8888", "filer2:8888", "filer3:8888"} + cluster := newTestCluster(hosts...) + + key := "lost-renewal-move-lock" + token1, _, primaryHost, err := cluster.acquireLock(key, "owner1", 30*time.Second) + require.NoError(t, err) + + // The renewal rotated the token; the client never learned token2. + token2, _, err := cluster.renewLock(key, "owner1", token1, 30*time.Second, primaryHost) + require.NoError(t, err) + require.NotEqual(t, token1, token2) + + _, backup := cluster.get(primaryHost).LockRing.GetPrimaryAndBackup(key) + require.NotEmpty(t, backup) + + // The backup copy must carry the superseded token too, or a promoted + // backup could not serve a retry of the lost renewal. + backupDlm := cluster.get(backup) + require.Eventually(t, func() bool { + lock, found := backupDlm.GetLock(key) + return found && lock.Token == token2 && lock.PrevToken == token1 + }, time.Second, time.Millisecond) + + // Primary dies; backup is promoted to primary. + cluster.removeNode(primaryHost) + require.True(t, backupDlm.PromoteLock(key)) + assert.Equal(t, backup, backupDlm.LockRing.GetPrimary(key)) + + // A retry with the superseded token recovers the current one. + _, replayed, _, movedTo, err := backupDlm.LockWithTimeout(key, time.Now().Add(30*time.Second).UnixNano(), token1, "owner1") + require.NoError(t, err) + assert.Empty(t, movedTo) + assert.Equal(t, token2, replayed) +} + func TestDLM_StaleReplicationRejected(t *testing.T) { // Verify that a stale replication (lower seq) does not overwrite a newer one lm := NewLockManager() // Insert backup with seq=3 - lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-new", "owner1", 1, 3) + lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-new", "", "owner1", 1, 3) lock, found := lm.GetLock("key1") require.True(t, found) assert.Equal(t, "token-new", lock.Token) assert.Equal(t, int64(3), lock.Seq) // Try to overwrite with stale seq=2 — should be rejected - lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-old", "owner1", 1, 2) + lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-old", "", "owner1", 1, 2) lock, found = lm.GetLock("key1") require.True(t, found) assert.Equal(t, "token-new", lock.Token, "stale replication should be rejected") assert.Equal(t, int64(3), lock.Seq) // Update with higher seq=4 — should succeed - lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-newer", "owner1", 1, 4) + lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-newer", "", "owner1", 1, 4) lock, found = lm.GetLock("key1") require.True(t, found) assert.Equal(t, "token-newer", lock.Token, "newer replication should be accepted") diff --git a/weed/cluster/lock_manager/lock_manager.go b/weed/cluster/lock_manager/lock_manager.go index e5c4a9d985..b9eb274485 100644 --- a/weed/cluster/lock_manager/lock_manager.go +++ b/weed/cluster/lock_manager/lock_manager.go @@ -24,6 +24,7 @@ type LockManager struct { } type Lock struct { Token string + PrevToken string // token superseded by the latest renewal; a holder whose renewal response was lost still presents it ExpiredAtNs int64 Key string // only used for moving locks Owner string @@ -102,19 +103,29 @@ func (lm *LockManager) Lock(path string, expiredAtNs int64, token string, owner generation = oldValue.Generation // keep same generation on renewal seq = oldValue.Seq + 1 glog.V(4).Infof("key %s old token %v owner %v => %v owner %v", path, oldValue.Token, oldValue.Owner, renewToken, owner) - lm.locks[path] = &Lock{Token: renewToken, ExpiredAtNs: expiredAtNs, Owner: owner, Generation: generation, Seq: seq} - return - } else { - if token == "" { - // new lock - glog.V(4).Infof("key %s locked by %v", path, oldValue.Owner) - err = fmt.Errorf("lock already owned by %v", oldValue.Owner) - return - } - glog.V(4).Infof("key %s expected token %v owner %v received %v from %v", path, oldValue.Token, oldValue.Owner, token, owner) - err = fmt.Errorf("lock: token mismatch") + lm.locks[path] = &Lock{Token: renewToken, PrevToken: oldValue.Token, ExpiredAtNs: expiredAtNs, Owner: owner, Generation: generation, Seq: seq} return } + if token != "" && token == oldValue.PrevToken && owner == oldValue.Owner { + // The latest renewal already rotated the token but its response + // was lost: replay the current token so the holder recovers it + // instead of conflicting with its own lease until it expires. + renewToken = oldValue.Token + generation = oldValue.Generation + seq = oldValue.Seq + 1 + oldValue.ExpiredAtNs = expiredAtNs + oldValue.Seq = seq + return + } + if token == "" { + // new lock + glog.V(4).Infof("key %s locked by %v", path, oldValue.Owner) + err = fmt.Errorf("lock already owned by %v", oldValue.Owner) + return + } + glog.V(4).Infof("key %s expected token %v owner %v received %v from %v", path, oldValue.Token, oldValue.Owner, token, owner) + err = fmt.Errorf("lock: token mismatch") + return } else { glog.V(4).Infof("key %s no lock owner %v", path, owner) if token == "" { @@ -148,7 +159,7 @@ func (lm *LockManager) Unlock(path string, token string) (isUnlocked bool, gener delete(lm.locks, path) return } - if oldValue.Token == token { + if oldValue.Token == token || (token != "" && token == oldValue.PrevToken) { isUnlocked = true generation = oldValue.Generation seq = oldValue.Seq + 1 @@ -210,7 +221,7 @@ func (lm *LockManager) SelectLocks(selectFn func(key string) bool) (locks []*Loc } // InsertLock inserts a lock from a transfer if it is newer than the current state. -func (lm *LockManager) InsertLock(path string, expiredAtNs int64, token string, owner string, generation int64, seq int64) bool { +func (lm *LockManager) InsertLock(path string, expiredAtNs int64, token string, prevToken string, owner string, generation int64, seq int64) bool { lm.accessLock.Lock() defer lm.accessLock.Unlock() @@ -220,7 +231,7 @@ func (lm *LockManager) InsertLock(path string, expiredAtNs int64, token string, return false } } - lm.locks[path] = &Lock{Token: token, ExpiredAtNs: expiredAtNs, Owner: owner, Generation: generation, Seq: seq} + lm.locks[path] = &Lock{Token: token, PrevToken: prevToken, ExpiredAtNs: expiredAtNs, Owner: owner, Generation: generation, Seq: seq} lm.advanceGenerationFloor(generation) return true } @@ -229,7 +240,7 @@ func (lm *LockManager) InsertLock(path string, expiredAtNs int64, token string, // It rejects stale mutations by comparing (generation, seq). If a current primary // already exists on this node, newer replicated state refreshes that primary copy // without demoting it back to a backup. -func (lm *LockManager) InsertBackupLock(path string, expiredAtNs int64, token string, owner string, generation int64, seq int64) { +func (lm *LockManager) InsertBackupLock(path string, expiredAtNs int64, token string, prevToken string, owner string, generation int64, seq int64) { lm.accessLock.Lock() defer lm.accessLock.Unlock() @@ -240,6 +251,7 @@ func (lm *LockManager) InsertBackupLock(path string, expiredAtNs int64, token st } lm.locks[path] = &Lock{ Token: token, + PrevToken: prevToken, ExpiredAtNs: expiredAtNs, Owner: owner, IsBackup: existing.IsBackup, @@ -249,7 +261,7 @@ func (lm *LockManager) InsertBackupLock(path string, expiredAtNs int64, token st lm.advanceGenerationFloor(generation) return } - lm.locks[path] = &Lock{Token: token, ExpiredAtNs: expiredAtNs, Owner: owner, IsBackup: true, Generation: generation, Seq: seq} + lm.locks[path] = &Lock{Token: token, PrevToken: prevToken, ExpiredAtNs: expiredAtNs, Owner: owner, IsBackup: true, Generation: generation, Seq: seq} lm.advanceGenerationFloor(generation) } diff --git a/weed/cluster/lock_manager/lock_manager_test.go b/weed/cluster/lock_manager/lock_manager_test.go index 5caf31da29..b0994e23fd 100644 --- a/weed/cluster/lock_manager/lock_manager_test.go +++ b/weed/cluster/lock_manager/lock_manager_test.go @@ -10,7 +10,7 @@ import ( func TestLockManager_GetLockOwnerIgnoresExpiredLock(t *testing.T) { lm := NewLockManager() - lm.InsertLock("key1", time.Now().Add(-time.Second).UnixNano(), "token1", "owner1", 7, 3) + lm.InsertLock("key1", time.Now().Add(-time.Second).UnixNano(), "token1", "", "owner1", 7, 3) owner, err := lm.GetLockOwner("key1") assert.Empty(t, owner) @@ -20,8 +20,8 @@ func TestLockManager_GetLockOwnerIgnoresExpiredLock(t *testing.T) { func TestLockManager_InsertLockRejectsStaleTransfer(t *testing.T) { lm := NewLockManager() - lm.InsertLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-new", "owner1", 8, 4) - lm.InsertLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-old", "owner1", 7, 3) + lm.InsertLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-new", "", "owner1", 8, 4) + lm.InsertLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-old", "", "owner1", 7, 3) lock, found := lm.GetLock("key1") assert.True(t, found) @@ -33,7 +33,7 @@ func TestLockManager_InsertLockRejectsStaleTransfer(t *testing.T) { func TestLockManager_InsertLockAdvancesGenerationCounter(t *testing.T) { lm := NewLockManager() - lm.InsertLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token1", "owner1", 12, 1) + lm.InsertLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token1", "", "owner1", 12, 1) _, renewToken, generation, _, err := lm.Lock("key2", time.Now().Add(30*time.Second).UnixNano(), "", "owner2") assert.NoError(t, err) @@ -44,8 +44,8 @@ func TestLockManager_InsertLockAdvancesGenerationCounter(t *testing.T) { func TestLockManager_InsertBackupLockRejectsOlderGeneration(t *testing.T) { lm := NewLockManager() - lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-new", "owner1", 8, 1) - lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-old", "owner1", 7, 9) + lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-new", "", "owner1", 8, 1) + lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-old", "", "owner1", 7, 9) lock, found := lm.GetLock("key1") assert.True(t, found) @@ -57,10 +57,10 @@ func TestLockManager_InsertBackupLockRejectsOlderGeneration(t *testing.T) { func TestLockManager_InsertBackupLockKeepsPrimaryRole(t *testing.T) { lm := NewLockManager() - ok := lm.InsertLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-old", "owner1", 8, 1) + ok := lm.InsertLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-old", "", "owner1", 8, 1) assert.True(t, ok) - lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-new", "owner1", 8, 2) + lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-new", "", "owner1", 8, 2) lock, found := lm.GetLock("key1") assert.True(t, found) @@ -73,7 +73,7 @@ func TestLockManager_InsertBackupLockKeepsPrimaryRole(t *testing.T) { func TestLockManager_RemoveBackupLockRejectsOlderGeneration(t *testing.T) { lm := NewLockManager() - lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-new", "owner1", 8, 1) + lm.InsertBackupLock("key1", time.Now().Add(30*time.Second).UnixNano(), "token-new", "", "owner1", 8, 1) removed := lm.RemoveBackupLockIfSeq("key1", 7, 9) assert.False(t, removed) @@ -84,3 +84,90 @@ func TestLockManager_RemoveBackupLockRejectsOlderGeneration(t *testing.T) { assert.Equal(t, int64(8), lock.Generation) assert.Equal(t, int64(1), lock.Seq) } + +func TestLockManager_LostRenewalResponseReplaysToken(t *testing.T) { + lm := NewLockManager() + expires := time.Now().Add(30 * time.Second).UnixNano() + + _, token1, _, _, err := lm.Lock("key1", expires, "", "owner1") + assert.NoError(t, err) + + // The renewal rotated token1 to token2; the client never learned it. + _, token2, _, _, err := lm.Lock("key1", expires, token1, "owner1") + assert.NoError(t, err) + assert.NotEqual(t, token1, token2) + + // Retrying with the superseded token replays the current one instead of + // conflicting with its own lease until it expires. + owner, replayed, _, _, err := lm.Lock("key1", expires, token1, "owner1") + assert.NoError(t, err) + assert.Equal(t, "owner1", owner) + assert.Equal(t, token2, replayed) + + // A replay from a different owner still mismatches. + _, _, _, _, err = lm.Lock("key1", expires, token1, "intruder") + assert.Error(t, err) + + // The next rotation retires the superseded token for good. + _, token3, _, _, err := lm.Lock("key1", expires, token2, "owner1") + assert.NoError(t, err) + assert.NotEqual(t, token2, token3) + _, _, _, _, err = lm.Lock("key1", expires, token1, "owner1") + assert.Error(t, err) + + // A superseded token still releases its own lock. + unlocked, _, _, err := lm.Unlock("key1", token2) + assert.NoError(t, err) + assert.True(t, unlocked) +} + +func TestLockManager_LostRenewalReplayAdvancesSeq(t *testing.T) { + lm := NewLockManager() + expires := time.Now().Add(30 * time.Second).UnixNano() + later := time.Now().Add(60 * time.Second).UnixNano() + + _, token1, _, _, err := lm.Lock("key1", expires, "", "owner1") + assert.NoError(t, err) + + _, token2, _, seq2, err := lm.Lock("key1", expires, token1, "owner1") + assert.NoError(t, err) + + // A replay that extends the lease must advance seq, otherwise backups + // reject the replicated expiry as a stale mutation and can grant the + // lock when the old expiry passes after a promotion. + _, replayed, _, seq3, err := lm.Lock("key1", later, token1, "owner1") + assert.NoError(t, err) + assert.Equal(t, token2, replayed) + assert.Equal(t, seq2+1, seq3) + + lock, found := lm.GetLock("key1") + assert.True(t, found) + assert.Equal(t, later, lock.ExpiredAtNs) + assert.Equal(t, seq3, lock.Seq) + + // Repeated retries keep advancing instead of colliding at one seq. + _, _, _, seq4, err := lm.Lock("key1", later, token1, "owner1") + assert.NoError(t, err) + assert.Equal(t, seq3+1, seq4) +} + +func TestLockManager_InsertBackupLockKeepsPrevToken(t *testing.T) { + lm := NewLockManager() + expires := time.Now().Add(30 * time.Second).UnixNano() + + // Backup copy of a renewed lock carries both the current token and the + // superseded one a client may still present after a lost response. + lm.InsertBackupLock("key1", expires, "token2", "token1", "owner1", 8, 2) + + lock, found := lm.GetLock("key1") + assert.True(t, found) + assert.Equal(t, "token2", lock.Token) + assert.Equal(t, "token1", lock.PrevToken) + + // Once promoted, the node still replays a retry of the superseded token. + assert.True(t, lm.PromoteLock("key1")) + owner, replayed, _, _, err := lm.Lock("key1", expires, "token1", "owner1") + assert.NoError(t, err) + assert.Equal(t, "owner1", owner) + assert.Equal(t, "token2", replayed) +} diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index b4024c9291..a33fc936cc 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -873,6 +873,7 @@ message Lock { int64 generation = 5; bool is_backup = 6; int64 seq = 7; + string prev_renew_token = 8; } message TransferLocksRequest { repeated Lock locks = 1; @@ -887,6 +888,7 @@ message ReplicateLockRequest { int64 generation = 5; bool is_unlock = 6; int64 seq = 7; + string prev_renew_token = 8; } message ReplicateLockResponse { } diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 163d9507eb..367cb5a6ca 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -1,7 +1,7 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: // protoc-gen-go v1.36.6 -// protoc v7.35.0 +// protoc v3.21.12 // source: filer.proto package filer_pb @@ -5935,16 +5935,17 @@ func (x *FindLockOwnerResponse) GetOwner() string { } type Lock struct { - state protoimpl.MessageState `protogen:"open.v1"` - Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` - RenewToken string `protobuf:"bytes,2,opt,name=renew_token,json=renewToken,proto3" json:"renew_token,omitempty"` - ExpiredAtNs int64 `protobuf:"varint,3,opt,name=expired_at_ns,json=expiredAtNs,proto3" json:"expired_at_ns,omitempty"` - Owner string `protobuf:"bytes,4,opt,name=owner,proto3" json:"owner,omitempty"` - Generation int64 `protobuf:"varint,5,opt,name=generation,proto3" json:"generation,omitempty"` - IsBackup bool `protobuf:"varint,6,opt,name=is_backup,json=isBackup,proto3" json:"is_backup,omitempty"` - Seq int64 `protobuf:"varint,7,opt,name=seq,proto3" json:"seq,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` + RenewToken string `protobuf:"bytes,2,opt,name=renew_token,json=renewToken,proto3" json:"renew_token,omitempty"` + ExpiredAtNs int64 `protobuf:"varint,3,opt,name=expired_at_ns,json=expiredAtNs,proto3" json:"expired_at_ns,omitempty"` + Owner string `protobuf:"bytes,4,opt,name=owner,proto3" json:"owner,omitempty"` + Generation int64 `protobuf:"varint,5,opt,name=generation,proto3" json:"generation,omitempty"` + IsBackup bool `protobuf:"varint,6,opt,name=is_backup,json=isBackup,proto3" json:"is_backup,omitempty"` + Seq int64 `protobuf:"varint,7,opt,name=seq,proto3" json:"seq,omitempty"` + PrevRenewToken string `protobuf:"bytes,8,opt,name=prev_renew_token,json=prevRenewToken,proto3" json:"prev_renew_token,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Lock) Reset() { @@ -6026,6 +6027,13 @@ func (x *Lock) GetSeq() int64 { return 0 } +func (x *Lock) GetPrevRenewToken() string { + if x != nil { + return x.PrevRenewToken + } + return "" +} + type TransferLocksRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Locks []*Lock `protobuf:"bytes,1,rep,name=locks,proto3" json:"locks,omitempty"` @@ -6107,16 +6115,17 @@ func (*TransferLocksResponse) Descriptor() ([]byte, []int) { } type ReplicateLockRequest struct { - state protoimpl.MessageState `protogen:"open.v1"` - Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` - RenewToken string `protobuf:"bytes,2,opt,name=renew_token,json=renewToken,proto3" json:"renew_token,omitempty"` - ExpiredAtNs int64 `protobuf:"varint,3,opt,name=expired_at_ns,json=expiredAtNs,proto3" json:"expired_at_ns,omitempty"` - Owner string `protobuf:"bytes,4,opt,name=owner,proto3" json:"owner,omitempty"` - Generation int64 `protobuf:"varint,5,opt,name=generation,proto3" json:"generation,omitempty"` - IsUnlock bool `protobuf:"varint,6,opt,name=is_unlock,json=isUnlock,proto3" json:"is_unlock,omitempty"` - Seq int64 `protobuf:"varint,7,opt,name=seq,proto3" json:"seq,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` + RenewToken string `protobuf:"bytes,2,opt,name=renew_token,json=renewToken,proto3" json:"renew_token,omitempty"` + ExpiredAtNs int64 `protobuf:"varint,3,opt,name=expired_at_ns,json=expiredAtNs,proto3" json:"expired_at_ns,omitempty"` + Owner string `protobuf:"bytes,4,opt,name=owner,proto3" json:"owner,omitempty"` + Generation int64 `protobuf:"varint,5,opt,name=generation,proto3" json:"generation,omitempty"` + IsUnlock bool `protobuf:"varint,6,opt,name=is_unlock,json=isUnlock,proto3" json:"is_unlock,omitempty"` + Seq int64 `protobuf:"varint,7,opt,name=seq,proto3" json:"seq,omitempty"` + PrevRenewToken string `protobuf:"bytes,8,opt,name=prev_renew_token,json=prevRenewToken,proto3" json:"prev_renew_token,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ReplicateLockRequest) Reset() { @@ -6198,6 +6207,13 @@ func (x *ReplicateLockRequest) GetSeq() int64 { return 0 } +func (x *ReplicateLockRequest) GetPrevRenewToken() string { + if x != nil { + return x.PrevRenewToken + } + return "" +} + type ReplicateLockResponse struct { state protoimpl.MessageState `protogen:"open.v1"` unknownFields protoimpl.UnknownFields @@ -7886,7 +7902,7 @@ const file_filer_proto_rawDesc = "" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x19\n" + "\bis_moved\x18\x02 \x01(\bR\aisMoved\"-\n" + "\x15FindLockOwnerResponse\x12\x14\n" + - "\x05owner\x18\x01 \x01(\tR\x05owner\"\xc4\x01\n" + + "\x05owner\x18\x01 \x01(\tR\x05owner\"\xee\x01\n" + "\x04Lock\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x1f\n" + "\vrenew_token\x18\x02 \x01(\tR\n" + @@ -7897,10 +7913,11 @@ const file_filer_proto_rawDesc = "" + "generation\x18\x05 \x01(\x03R\n" + "generation\x12\x1b\n" + "\tis_backup\x18\x06 \x01(\bR\bisBackup\x12\x10\n" + - "\x03seq\x18\a \x01(\x03R\x03seq\"<\n" + + "\x03seq\x18\a \x01(\x03R\x03seq\x12(\n" + + "\x10prev_renew_token\x18\b \x01(\tR\x0eprevRenewToken\"<\n" + "\x14TransferLocksRequest\x12$\n" + "\x05locks\x18\x01 \x03(\v2\x0e.filer_pb.LockR\x05locks\"\x17\n" + - "\x15TransferLocksResponse\"\xd4\x01\n" + + "\x15TransferLocksResponse\"\xfe\x01\n" + "\x14ReplicateLockRequest\x12\x12\n" + "\x04name\x18\x01 \x01(\tR\x04name\x12\x1f\n" + "\vrenew_token\x18\x02 \x01(\tR\n" + @@ -7911,7 +7928,8 @@ const file_filer_proto_rawDesc = "" + "generation\x18\x05 \x01(\x03R\n" + "generation\x12\x1b\n" + "\tis_unlock\x18\x06 \x01(\bR\bisUnlock\x12\x10\n" + - "\x03seq\x18\a \x01(\x03R\x03seq\"\x17\n" + + "\x03seq\x18\a \x01(\x03R\x03seq\x12(\n" + + "\x10prev_renew_token\x18\b \x01(\tR\x0eprevRenewToken\"\x17\n" + "\x15ReplicateLockResponse\"\xe6\x02\n" + "\x18StreamMutateEntryRequest\x12\x1d\n" + "\n" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 5c5becb7bb..aba514e8be 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -5615,6 +5615,13 @@ func (m *Lock) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if len(m.PrevRenewToken) > 0 { + i -= len(m.PrevRenewToken) + copy(dAtA[i:], m.PrevRenewToken) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.PrevRenewToken))) + i-- + dAtA[i] = 0x42 + } if m.Seq != 0 { i = protohelpers.EncodeVarint(dAtA, i, uint64(m.Seq)) i-- @@ -5772,6 +5779,13 @@ func (m *ReplicateLockRequest) MarshalToSizedBufferVT(dAtA []byte) (int, error) i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if len(m.PrevRenewToken) > 0 { + i -= len(m.PrevRenewToken) + copy(dAtA[i:], m.PrevRenewToken) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(m.PrevRenewToken))) + i-- + dAtA[i] = 0x42 + } if m.Seq != 0 { i = protohelpers.EncodeVarint(dAtA, i, uint64(m.Seq)) i-- @@ -8843,6 +8857,10 @@ func (m *Lock) SizeVT() (n int) { if m.Seq != 0 { n += 1 + protohelpers.SizeOfVarint(uint64(m.Seq)) } + l = len(m.PrevRenewToken) + if l > 0 { + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } n += len(m.unknownFields) return n } @@ -8903,6 +8921,10 @@ func (m *ReplicateLockRequest) SizeVT() (n int) { if m.Seq != 0 { n += 1 + protohelpers.SizeOfVarint(uint64(m.Seq)) } + l = len(m.PrevRenewToken) + if l > 0 { + n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) + } n += len(m.unknownFields) return n } @@ -24263,6 +24285,38 @@ func (m *Lock) UnmarshalVT(dAtA []byte) error { break } } + case 8: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field PrevRenewToken", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.PrevRenewToken = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) @@ -24623,6 +24677,38 @@ func (m *ReplicateLockRequest) UnmarshalVT(dAtA []byte) error { break } } + case 8: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field PrevRenewToken", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + intStringLen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.PrevRenewToken = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/server/filer_grpc_server_dlm.go b/weed/server/filer_grpc_server_dlm.go index dc2864818d..b6516dcf54 100644 --- a/weed/server/filer_grpc_server_dlm.go +++ b/weed/server/filer_grpc_server_dlm.go @@ -152,7 +152,7 @@ func (fs *FilerServer) FindLockOwner(ctx context.Context, req *filer_pb.FindLock func (fs *FilerServer) TransferLocks(ctx context.Context, req *filer_pb.TransferLocksRequest) (*filer_pb.TransferLocksResponse, error) { for _, lock := range req.Locks { - fs.filer.Dlm.InsertLock(lock.Name, lock.ExpiredAtNs, lock.RenewToken, lock.Owner, lock.Generation, lock.Seq) + fs.filer.Dlm.InsertLock(lock.Name, lock.ExpiredAtNs, lock.RenewToken, lock.PrevRenewToken, lock.Owner, lock.Generation, lock.Seq) } return &filer_pb.TransferLocksResponse{}, nil @@ -166,7 +166,7 @@ func (fs *FilerServer) ReplicateLock(ctx context.Context, req *filer_pb.Replicat fs.filer.Dlm.RemoveBackupLockIfSeq(req.Name, req.Generation, req.Seq) glog.V(4).Infof("FILER REPLICATE: removed backup lock %s generation=%d seq=%d", req.Name, req.Generation, req.Seq) } else { - fs.filer.Dlm.InsertBackupLock(req.Name, req.ExpiredAtNs, req.RenewToken, req.Owner, req.Generation, req.Seq) + fs.filer.Dlm.InsertBackupLock(req.Name, req.ExpiredAtNs, req.RenewToken, req.PrevRenewToken, req.Owner, req.Generation, req.Seq) glog.V(4).Infof("FILER REPLICATE: inserted backup lock %s owner=%s generation=%d seq=%d", req.Name, req.Owner, req.Generation, req.Seq) } return &filer_pb.ReplicateLockResponse{}, nil @@ -194,7 +194,7 @@ func (fs *FilerServer) OnDlmChangeSnapshot(snapshot []pb.ServerAddress) { } // Replicate to (possibly new) backup if backup != "" && fs.filer.Dlm.ReplicateFn != nil { - go fs.filer.Dlm.ReplicateFn(backup, lock.Key, lock.ExpiredAtNs, lock.Token, lock.Owner, lock.Generation, lock.Seq, false) + go fs.filer.Dlm.ReplicateFn(backup, lock.Key, lock.ExpiredAtNs, lock.Token, lock.PrevToken, lock.Owner, lock.Generation, lock.Seq, false) } } else if backup == fs.option.Host { if !lock.IsBackup { @@ -205,12 +205,13 @@ func (fs *FilerServer) OnDlmChangeSnapshot(snapshot []pb.ServerAddress) { _, err := client.TransferLocks(ctx, &filer_pb.TransferLocksRequest{ Locks: []*filer_pb.Lock{ { - Name: lock.Key, - RenewToken: lock.Token, - ExpiredAtNs: lock.ExpiredAtNs, - Owner: lock.Owner, - Generation: lock.Generation, - Seq: lock.Seq, + Name: lock.Key, + RenewToken: lock.Token, + PrevRenewToken: lock.PrevToken, + ExpiredAtNs: lock.ExpiredAtNs, + Owner: lock.Owner, + Generation: lock.Generation, + Seq: lock.Seq, }, }, }) @@ -233,12 +234,13 @@ func (fs *FilerServer) OnDlmChangeSnapshot(snapshot []pb.ServerAddress) { _, err := client.TransferLocks(ctx, &filer_pb.TransferLocksRequest{ Locks: []*filer_pb.Lock{ { - Name: lock.Key, - RenewToken: lock.Token, - ExpiredAtNs: lock.ExpiredAtNs, - Owner: lock.Owner, - Generation: lock.Generation, - Seq: lock.Seq, + Name: lock.Key, + RenewToken: lock.Token, + PrevRenewToken: lock.PrevToken, + ExpiredAtNs: lock.ExpiredAtNs, + Owner: lock.Owner, + Generation: lock.Generation, + Seq: lock.Seq, }, }, }) @@ -257,18 +259,19 @@ func (fs *FilerServer) OnDlmChangeSnapshot(snapshot []pb.ServerAddress) { // SetupDlmReplication sets up the replication callback for the DLM. // Called during filer server initialization. func (fs *FilerServer) SetupDlmReplication() { - fs.filer.Dlm.ReplicateFn = func(server pb.ServerAddress, key string, expiredAtNs int64, token string, owner string, generation int64, seq int64, isUnlock bool) { + fs.filer.Dlm.ReplicateFn = func(server pb.ServerAddress, key string, expiredAtNs int64, token string, prevToken string, owner string, generation int64, seq int64, isUnlock bool) { ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() err := pb.WithFilerClient(false, 0, server, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { _, err := client.ReplicateLock(ctx, &filer_pb.ReplicateLockRequest{ - Name: key, - RenewToken: token, - ExpiredAtNs: expiredAtNs, - Owner: owner, - Generation: generation, - Seq: seq, - IsUnlock: isUnlock, + Name: key, + RenewToken: token, + PrevRenewToken: prevToken, + ExpiredAtNs: expiredAtNs, + Owner: owner, + Generation: generation, + Seq: seq, + IsUnlock: isUnlock, }) return err }) diff --git a/weed/server/filer_grpc_server_dlm_test.go b/weed/server/filer_grpc_server_dlm_test.go index 7c82e9faac..c5cf6478e0 100644 --- a/weed/server/filer_grpc_server_dlm_test.go +++ b/weed/server/filer_grpc_server_dlm_test.go @@ -22,7 +22,7 @@ func TestFindLockOwnerExpiredLockReturnsNotFound(t *testing.T) { }, } fs.filer.Dlm.LockRing.SetSnapshot([]pb.ServerAddress{"filer1:8888"}, 0) - fs.filer.Dlm.InsertLock("expired-lock", time.Now().Add(-time.Second).UnixNano(), "token1", "owner1", 5, 2) + fs.filer.Dlm.InsertLock("expired-lock", time.Now().Add(-time.Second).UnixNano(), "token1", "", "owner1", 5, 2) resp, err := fs.FindLockOwner(context.Background(), &filer_pb.FindLockOwnerRequest{ Name: "expired-lock",