Make lock cache expirable

internal/dcs/config.go

@@ -11,6 +11,7 @@ import (
 type ZookeeperConfig struct {
 	Hostname              string                   `config:"hostname" yaml:"hostname"`
 	SessionTimeout        time.Duration            `config:"session_timeout" yaml:"session_timeout"`
+	LockHeldTTL           time.Duration            `config:"lock_held_ttl" yaml:"lock_held_ttl"`
 	Namespace             string                   `config:"namespace,required" yaml:"namespace"`
 	Hosts                 []string                 `config:"hosts,required" yaml:"hosts"`
 	BackoffInterval       time.Duration            `config:"backoff_interval" yaml:"backoff_interval"`
@@ -57,6 +58,7 @@ func DefaultZookeeperConfig() (ZookeeperConfig, error) {
 	config := ZookeeperConfig{
 		Hostname:              hostname,
 		SessionTimeout:        2 * time.Second,
+		LockHeldTTL:           30 * time.Second,
 		BackoffInterval:       backoff.DefaultInitialInterval,
 		BackoffRandFactor:     backoff.DefaultRandomizationFactor,
 		BackoffMultiplier:     backoff.DefaultMultiplier,

internal/dcs/zk.go

@@ -340,9 +340,11 @@ func (z *zkDCS) retryDelete(path string, version int32) (err error) {
 
 func (z *zkDCS) AcquireLock(path string) bool {
 	fullPath := z.buildFullPath(path)
-	_, hasLock := z.lockHeld.Load(fullPath)
-	if hasLock {
-		return true
+	if cached, hasLock := z.lockHeld.Load(fullPath); hasLock {
+		if time.Since(cached.(time.Time)) < z.config.LockHeldTTL {
+			return true
+		}
+		z.lockHeld.Delete(fullPath)
 	}
 	self := z.getSelfLockOwner()
 	data, _, err := z.retryGet(fullPath)
@@ -362,7 +364,7 @@ func (z *zkDCS) AcquireLock(path string) bool {
 			}
 			return false
 		}
-		z.lockHeld.Store(fullPath, struct{}{})
+		z.lockHeld.Store(fullPath, time.Now())
 		return true
 	}
 	owner := LockOwner{}
@@ -371,7 +373,7 @@ func (z *zkDCS) AcquireLock(path string) bool {
 		return false
 	}
 	if owner == self {
-		z.lockHeld.Store(fullPath, struct{}{})
+		z.lockHeld.Store(fullPath, time.Now())
 		return true
 	}
 	return false