seaweedfs/weed/cluster/lock_client.go

package cluster

import (
	"context"
	"fmt"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/seaweedfs/seaweedfs/weed/cluster/lock_manager"
	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/pb"
	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"google.golang.org/grpc"
)

type LockClient struct {
	grpcDialOption  grpc.DialOption
	maxLockDuration time.Duration
	sleepDuration   time.Duration
	seedFiler       pb.ServerAddress

	// ring is an optional client-side view of the filer lock hash ring. When
	// populated, a new lock starts at the key's primary filer instead of the
	// seed filer, avoiding the seed->primary forward hop. A stale view stays
	// correct: the filer forwards to the real primary as a fallback.
	ringMu      sync.RWMutex
	ring        *lock_manager.HashRing
	ringServers []pb.ServerAddress
	ringVersion int64

	// priorRing is the ring before the most recent change, kept for priorWindow so a
	// route-by-key caller can consult a just-moved key's previous owner during a
	// rebalance. Mirrors the master's lock_manager.LockRing.PriorOwner cooling-off.
	priorRing     *lock_manager.HashRing
	ringChangedAt time.Time
	priorWindow   time.Duration

	// noLockServerRetryPeriod bounds retries when every filer reports "no
	// lock server found": an empty lock ring is a systemic fault that waiting
	// on a lock holder cannot resolve, unlike ordinary contention. The bound
	// is long enough to ride out a master leader change (ring reset + fresh
	// 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 {
	return &LockClient{
		grpcDialOption:          grpcDialOption,
		maxLockDuration:         5 * time.Second,
		sleepDuration:           2473 * time.Millisecond,
		seedFiler:               seedFiler,
		priorWindow:             5 * time.Second,
		noLockServerRetryPeriod: 15 * time.Second,
		lockRpcTimeout:          10 * time.Second,
	}
}

// SetRing mirrors the master's LockRingUpdate so the client computes the same
// primary the filers do. A non-zero version at or below the current one is
// ignored once a ring exists, dropping reordered and redundant broadcasts;
// version 0 always applies (bootstrap).
func (lc *LockClient) SetRing(servers []pb.ServerAddress, version int64) {
	lc.ringMu.Lock()
	defer lc.ringMu.Unlock()
	if version != 0 && version <= lc.ringVersion && lc.ring != nil {
		return
	}
	lc.ringVersion = version
	sorted := slices.Clone(servers)
	slices.Sort(sorted)
	if slices.Equal(sorted, lc.ringServers) {
		return
	}
	lc.ringServers = sorted
	// Build a fresh ring (not an in-place mutation) so the outgoing ring survives as
	// priorRing with its own servers for the cooling-off window.
	newRing := lock_manager.NewHashRing(lock_manager.DefaultVnodeCount)
	newRing.SetServers(sorted)
	if lc.ring != nil {
		lc.priorRing = lc.ring
		lc.ringChangedAt = time.Now()
	}
	lc.ring = newRing
}

// ResetRing clears only the version gate so the first update from a
// different master applies unconditionally: ring versions are per-master
// monotonic and a high version accepted from a former leader must not
// reject the new leader's snapshot. The last ring keeps routing during the
// gap rather than falling back to the seed filer, and the arriving ring
// becomes the prior ring for the cooling-off window.
func (lc *LockClient) ResetRing() {
	lc.ringMu.Lock()
	defer lc.ringMu.Unlock()
	lc.ringVersion = 0
}

// hostForKey returns the filer that should own key per the current ring view,
// falling back to the seed filer when no view has been received yet.
func (lc *LockClient) hostForKey(key string) pb.ServerAddress {
	lc.ringMu.RLock()
	defer lc.ringMu.RUnlock()
	if lc.ring == nil {
		return lc.seedFiler
	}
	if primary := lc.ring.GetPrimary(key); primary != "" {
		return primary
	}
	return lc.seedFiler
}

// PrimaryForKey returns the ring owner for key, or "" before any ring arrives.
// Unlike hostForKey it does not fall back to the seed, so a route-by-key caller
// stays on the distributed lock until the ring is known.
func (lc *LockClient) PrimaryForKey(key string) pb.ServerAddress {
	lc.ringMu.RLock()
	defer lc.ringMu.RUnlock()
	if lc.ring == nil {
		return ""
	}
	return lc.ring.GetPrimary(key)
}

// PriorOwnerForKey returns key's owner from the prior ring while a rebalance is
// within the cooling-off window and ownership actually moved, else "". Lets a
// route-by-key reader consult a just-moved key's previous owner before the new
// owner's NotFound is final (the new owner may not have replicated the key yet).
func (lc *LockClient) PriorOwnerForKey(key string) pb.ServerAddress {
	lc.ringMu.RLock()
	defer lc.ringMu.RUnlock()
	if lc.ring == nil || lc.priorRing == nil {
		return ""
	}
	if time.Since(lc.ringChangedAt) > lc.priorWindow {
		return ""
	}
	current := lc.ring.GetPrimary(key)
	prior := lc.priorRing.GetPrimary(key)
	if prior != "" && prior != current {
		return prior
	}
	return ""
}

type LiveLock struct {
	generation          int64 // fencing token from the lock server; MUST stay first so the 64-bit atomic ops stay 8-byte aligned on 32-bit ARM
	key                 string
	renewToken          string
	expireAtNs          int64
	hostFiler           pb.ServerAddress
	cancelCh            chan struct{}
	renewalDone         chan struct{} // closed when the renewal goroutine exits; nil if there is none
	grpcDialOption      grpc.DialOption
	isLocked            int32 // 0 = unlocked, 1 = locked; use atomic operations
	self                string
	lc                  *LockClient
	owner               string
	lockTTL             time.Duration
	consecutiveFailures int // Track connection failures to trigger fallback
}

// NewShortLivedLock creates a lock with a 5-second duration.
// It returns nil when the lock cannot be acquired because no lock server
// exists; ordinary contention is still waited out.
func (lc *LockClient) NewShortLivedLock(key string, owner string) (lock *LiveLock) {
	lock = &LiveLock{
		key:            key,
		hostFiler:      lc.hostForKey(key),
		cancelCh:       make(chan struct{}),
		expireAtNs:     time.Now().Add(5 * time.Second).UnixNano(),
		grpcDialOption: lc.grpcDialOption,
		self:           owner,
		lc:             lc,
	}
	if err := lock.retryUntilLocked(5 * time.Second); err != nil {
		glog.Warningf("create lock %s: %v", key, err)
		return nil
	}
	return
}

// NewBlockingLongLivedLock blocks until the lock is acquired, then starts a
// background renewal goroutine that keeps the lock alive. This combines the
// synchronous acquisition of NewShortLivedLock with the auto-renewal of
// StartLongLivedLock. Release with Stop().
func (lc *LockClient) NewBlockingLongLivedLock(key, owner string, lockTTL time.Duration) *LiveLock {
	if lockTTL == 0 {
		lockTTL = lock_manager.LiveLockTTL
	}
	lock := &LiveLock{
		key:            key,
		hostFiler:      lc.hostForKey(key),
		cancelCh:       make(chan struct{}),
		expireAtNs:     time.Now().Add(lockTTL).UnixNano(),
		grpcDialOption: lc.grpcDialOption,
		self:           owner,
		lc:             lc,
		lockTTL:        lockTTL,
	}
	// Block until acquired
	if err := lock.retryUntilLocked(lockTTL); err != nil {
		glog.Warningf("create lock %s: %v", key, err)
		return nil
	}
	// Start renewal goroutine using a ticker for interruptible sleep
	lock.renewalDone = make(chan struct{})
	go func() {
		defer close(lock.renewalDone)
		renewInterval := lockTTL / 2
		ticker := time.NewTicker(renewInterval)
		defer ticker.Stop()
		for {
			select {
			case <-lock.cancelCh:
				return
			case <-ticker.C:
				if err := lock.AttemptToLock(lockTTL); err != nil {
					glog.V(0).Infof("lock renewal failed for %s: %v", key, err)
					atomic.StoreInt32(&lock.isLocked, 0)
				}
			}
		}
	}()
	return lock
}

// StartLongLivedLock starts a goroutine to lock the key and returns immediately.
// lockTTL specifies how long the lock should be held. The renewal interval is
// automatically derived as lockTTL / 2 to ensure timely renewals.
func (lc *LockClient) StartLongLivedLock(key string, owner string, onLockOwnerChange func(newLockOwner string), lockTTL time.Duration) (lock *LiveLock) {
	lock = &LiveLock{
		key:            key,
		hostFiler:      lc.hostForKey(key),
		cancelCh:       make(chan struct{}),
		expireAtNs:     time.Now().Add(lockTTL).UnixNano(),
		grpcDialOption: lc.grpcDialOption,
		self:           owner,
		lc:             lc,
		lockTTL:        lockTTL,
	}
	if lock.lockTTL == 0 {
		lock.lockTTL = lock_manager.LiveLockTTL
	}
	lock.renewalDone = make(chan struct{})
	go func() {
		defer close(lock.renewalDone)
		renewInterval := lock.lockTTL / 2
		isLocked := false
		lockOwner := ""
		for {
			// Check for cancellation BEFORE attempting to lock to avoid race condition
			// where Stop() is called after sleep but before lock attempt
			select {
			case <-lock.cancelCh:
				return
			default:
			}

			if isLocked {
				if err := lock.AttemptToLock(lock.lockTTL); err != nil {
					glog.V(0).Infof("Lost lock %s: %v", key, err)
					isLocked = false
					atomic.StoreInt32(&lock.isLocked, 0)
				}
			} else {
				if err := lock.AttemptToLock(lock.lockTTL); err == nil {
					isLocked = true
					// Note: AttemptToLock already sets lock.isLocked atomically on success
				}
			}
			if lockOwner != lock.LockOwner() && lock.LockOwner() != "" {
				glog.V(0).Infof("Lock owner changed from %s to %s", lockOwner, lock.LockOwner())
				onLockOwnerChange(lock.LockOwner())
				lockOwner = lock.LockOwner()
			}
			// Sleep until the next attempt, but wake immediately on Stop() so
			// the goroutine exits and closes renewalDone before Stop()'s bounded
			// wait elapses. An uninterruptible sleep here (up to 5*renewInterval
			// when unlocked) can outlast that wait and break the shutdown
			// synchronization.
			sleepFor := renewInterval
			if !isLocked {
				sleepFor = 5 * renewInterval
			}
			timer := time.NewTimer(sleepFor)
			select {
			case <-lock.cancelCh:
				timer.Stop()
				return
			case <-timer.C:
			}
		}
	}()
	return
}

// retryUntilLocked blocks until the lock is acquired, polling at the steady
// short cadence that AttemptToLock already enforces on contention (~1s). It
// deliberately avoids util.RetryUntil's exponential backoff (which grows to
// several seconds): when a holder on another mount releases the lock, the
// 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 {
	// 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 {
			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.Now()
			continue
		}
		// Anything else — "no lock server found", a dead ring member refusing
		// 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()
		}
	}
}

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, 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 false, err
	}
	if errorMessage != "" {
		if strings.Contains(errorMessage, "lock already owned") {
			glog.V(3).Infof("LOCK: doLock returned error message for key=%s: %s", lock.key, errorMessage)
		} else {
			glog.V(2).Infof("LOCK: doLock returned error message for key=%s: %s", lock.key, errorMessage)
		}
		time.Sleep(time.Second)
		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 true, nil
}

func (lock *LiveLock) StopShortLivedLock() error {
	if atomic.LoadInt32(&lock.isLocked) == 0 {
		return nil
	}
	defer func() {
		atomic.StoreInt32(&lock.isLocked, 0)
	}()
	return pb.WithFilerClient(false, 0, lock.hostFiler, lock.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
		ctx, cancel := context.WithTimeout(context.Background(), lock.lc.lockRpcTimeout)
		defer cancel()
		_, err := client.DistributedUnlock(ctx, &filer_pb.UnlockRequest{
			Name:       lock.key,
			RenewToken: lock.renewToken,
		})
		return err
	})
}

// Stop stops a long-lived lock by closing the cancel channel and releasing the lock
func (lock *LiveLock) Stop() error {
	// Close the cancel channel to stop the long-lived lock goroutine
	select {
	case <-lock.cancelCh:
		// Already closed
	default:
		close(lock.cancelCh)
	}

	// Wait for the renewal goroutine to fully exit before unlocking. A renewal
	// in flight when we close cancelCh rotates renewToken on the server; if we
	// then unlock with the token we read here, the unlock fails with a token
	// mismatch and the lock lingers until its TTL expires — blocking other
	// mounts waiting on the same file. Waiting for the goroutine to return also
	// makes the renewToken read below race-free (channel close = happens-before).
	if lock.renewalDone != nil {
		select {
		case <-lock.renewalDone:
		case <-time.After(lock.lockTTL + 2*time.Second):
			// The renewal goroutine is wedged, almost certainly in a stuck
			// renewal RPC. Do not unlock here: the renewToken may be rotated
			// when that RPC finally returns, so an unlock sent now could race
			// it, be rejected on a stale token, and leave the lock lingering
			// anyway. cancelCh is closed, so the goroutine stops renewing once
			// its in-flight call returns and the lock then expires within its
			// TTL on its own.
			glog.Warningf("lock %s: renewal goroutine still running at shutdown; letting lock expire via TTL", lock.key)
			return nil
		}
	}

	// Also release the lock if held
	// Note: We intentionally don't clear renewToken here because
	// StopShortLivedLock needs it to properly unlock
	return lock.StopShortLivedLock()
}

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)

	previousHostFiler := lock.hostFiler
	previousOwner := lock.owner

	err = pb.WithFilerClient(false, 0, lock.hostFiler, lock.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
		// 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,
			IsMoved:       false,
			Owner:         lock.self,
		})
		glog.V(4).Infof("LOCK: DistributedLock response - key=%s err=%v", lock.key, err)
		if err == nil && resp != nil {
			// 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 {
			// 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
			if resp.LockHostMovedTo != "" && !pb.ServerAddress(resp.LockHostMovedTo).Equals(previousHostFiler) {
				// Only log if the host actually changed
				glog.V(2).Infof("LOCK: Host changed from %s to %s for key=%s", previousHostFiler, resp.LockHostMovedTo, lock.key)
				lock.hostFiler = pb.ServerAddress(resp.LockHostMovedTo)
				// Don't update seedFiler - keep original for fallback
			} else if resp.LockHostMovedTo != "" {
				lock.hostFiler = pb.ServerAddress(resp.LockHostMovedTo)
			}
			if resp.LockOwner != "" && resp.LockOwner != previousOwner {
				// Only log if the owner actually changed
				glog.V(2).Infof("LOCK: Owner changed from %s to %s for key=%s", previousOwner, resp.LockOwner, lock.key)
				lock.owner = resp.LockOwner
			} else if resp.LockOwner != "" {
				lock.owner = resp.LockOwner
			} else if previousOwner != "" {
				glog.V(2).Infof("LOCK: Owner cleared for key=%s", lock.key)
				lock.owner = ""
			}
		}
		return err
	})

	if err != nil && !lock.hostFiler.Equals(lock.lc.seedFiler) {
		lock.consecutiveFailures++
		// Fall back to seed filer after 3 consecutive connection failures
		if lock.consecutiveFailures >= 3 {
			glog.V(0).Infof("LOCK: Connection failed %d times for key=%s filer=%s, falling back to seed filer=%s",
				lock.consecutiveFailures, lock.key, lock.hostFiler, lock.lc.seedFiler)
			lock.hostFiler = lock.lc.seedFiler
			lock.consecutiveFailures = 0
		}
	}

	return
}

func (lock *LiveLock) LockOwner() string {
	return lock.owner
}

// Generation returns the fencing token for this lock.
// It increments on each fresh acquisition and stays the same on renewal.
func (lock *LiveLock) Generation() int64 {
	return atomic.LoadInt64(&lock.generation)
}

// IsLocked returns true if this instance currently holds the lock
func (lock *LiveLock) IsLocked() bool {
	return atomic.LoadInt32(&lock.isLocked) == 1
}