Locks¶

etcd-recipes ships three mutual-exclusion recipes:

All three extend EtcdConnector, so they share the lifecycle, exception, and connection-state surface described in Core concepts.

The EtcdLock interface¶

DistributedMutex and the two views of DistributedReadWriteLock implement EtcdLock:

interface EtcdLock {
  fun lock()
  fun tryLock(timeout: Duration): Boolean
  fun tryLock(timeout: Long, timeUnit: TimeUnit): Boolean
  fun unlock(): Boolean
  val isHeldByCurrentThread: Boolean
  val isLocked: Boolean
  val holdCount: Int
  fun addLockLostListener(listener: LockLostListener)
  fun removeLockLostListener(listener: LockLostListener)
}

DistributedMutex¶

Built on etcd's native lock service, so ordering is server-side FIFO by revision.

DistributedMutex(client, "/locks/orders").use { mutex ->
  mutex.withLock {
    // Only one client across the cluster runs this at a time.
    logger.info { "Critical section" }
  }
}

try (DistributedMutex mutex = new DistributedMutex(client, "/locks/orders")) {
  mutex.lock();
  try {
    // Only one client across the cluster runs this at a time.
    System.out.println("Critical section");
  } finally {
    mutex.unlock();
  }
}

withLock { } is an inline Kotlin extension. It does compile to a static EtcdLockKt.withLock(lock, Function0) that Java can call, but a Java lambda would have to return Unit.INSTANCE, so it buys nothing — Java callers take the lock and release it in a finally. See the Java guide.

Acquiring with a timeout¶

Blocking forever is rarely what a service wants. tryLock bounds the wait:

DistributedMutex(client, "/locks/orders").use { mutex ->
  if (mutex.tryLock(5.seconds)) {
    try {
      logger.info { "Acquired within the timeout" }
    } finally {
      mutex.unlock()
    }
  } else {
    logger.info { "Someone else holds the lock; moving on" }
  }
}

try (DistributedMutex mutex = new DistributedMutex(client, "/locks/orders")) {
  // Java uses the (long, TimeUnit) overload; the Duration one is Kotlin-facing.
  if (mutex.tryLock(5, TimeUnit.SECONDS)) {
    try {
      System.out.println("Acquired within the timeout");
    } finally {
      mutex.unlock();
    }
  } else {
    System.out.println("Someone else holds the lock; moving on");
  }
}

Losing the lock¶

This is the part that has no single-JVM analogue, and it is the part worth reading twice.

A mutex hold is backed by a lease. If the process stalls or the network partitions long enough for that lease to expire, etcd has already handed the lock to the next waiter. The recipe does not try to heal that lease or reclaim the lock: doing so would race the new holder and defeat the entire point of the lock.

Instead, loss is cooperative. You are told, and you decide:

DistributedMutex(client, "/locks/orders").use { mutex ->
  // The acquisition lease is deliberately never healed. If it expires, etcd has
  // already promoted the next waiter, so the hold is gone and cannot be reclaimed.
  mutex.addLockLostListener { cause ->
    logger.warn { "Lost the lock, abandoning work: $cause" }
  }

  mutex.withLock {
    // unlock() returns false if the hold was lost while inside the section.
    logger.info { "Working" }
  }
}

try (DistributedMutex mutex = new DistributedMutex(client, "/locks/orders")) {
  // The acquisition lease is deliberately never healed. If it expires, etcd has
  // already promoted the next waiter, so the hold is gone and cannot be reclaimed.
  mutex.addLockLostListener(cause ->
    System.out.println("Lost the lock, abandoning work: " + cause));

  mutex.lock();
  try {
    System.out.println("Working");
  } finally {
    // Returns false if the hold was lost while inside the section.
    mutex.unlock();
  }
}

Three things happen on loss:

• every registered LockLostListener fires,
• connectionState moves to LOST,
• the eventual unlock() returns false rather than throwing.

If you would rather the holding thread be interrupted the instant the lock is lost — so it cannot keep mutating state it no longer owns — opt in with interruptOnLockLoss:

// Opt in to having the holding thread interrupted the moment the lock is lost,
// rather than letting it run on against state it no longer owns.
DistributedMutex(
  client = client,
  lockPath = "/locks/orders",
  leaseTtlSecs = 5L,
  interruptOnLockLoss = true,
).use { mutex ->
  mutex.withLock {
    logger.info { "Interrupted if the lease expires under us" }
  }
}

// Opt in to having the holding thread interrupted the moment the lock is lost,
// rather than letting it run on against state it no longer owns. Java has no
// named arguments, so every preceding parameter must be supplied positionally.
try (DistributedMutex mutex =
       new DistributedMutex(
         client,
         "/locks/orders",
         5L,                                  // leaseTtlSecs
         io.etcd.recipes.common.ResilienceConfig.DEFAULT,
         "worker-1",                          // clientId
         true)) {                             // interruptOnLockLoss
  mutex.lock();
  try {
    System.out.println("Interrupted if the lease expires under us");
  } finally {
    mutex.unlock();
  }
}

It defaults to false because interrupting a thread that is midway through non-idempotent work is not automatically safer than letting it finish and fail its own commit. Choose deliberately.

Reentrancy¶

Holds are per-thread and reentrant, tracked by holdCount:

DistributedMutex(client, "/locks/orders").use { mutex ->
  mutex.withLock {
    mutex.withLock {
      // Reentrant within the same thread; holdCount is now 2.
      logger.info { "holdCount=${mutex.holdCount}" }
    }
  }
}

DistributedReadWriteLock¶

Many concurrent readers, or exactly one writer. Fair: waiters are served FIFO by create revision, so a steady stream of readers cannot starve a waiting writer.

DistributedReadWriteLock(client, "/locks/catalog").use { rwLock ->
  // Any number of readers may hold the read lock at once...
  rwLock.readLock.withLock {
    logger.info { "Reading the catalog" }
  }

  // ...but a writer excludes every reader and every other writer.
  rwLock.writeLock.withLock {
    logger.info { "Rewriting the catalog" }
  }
}

try (DistributedReadWriteLock rwLock = new DistributedReadWriteLock(client, "/locks/catalog")) {
  // Any number of readers may hold the read lock at once...
  rwLock.getReadLock().lock();
  try {
    System.out.println("Reading the catalog");
  } finally {
    rwLock.getReadLock().unlock();
  }

  // ...but a writer excludes every reader and every other writer.
  rwLock.getWriteLock().lock();
  try {
    System.out.println("Rewriting the catalog");
  } finally {
    rwLock.getWriteLock().unlock();
  }
}

DistributedReadWriteLock does not itself implement EtcdLock — it exposes two properties that do, readLock and writeLock. Everything from the mutex section (timeouts, lock-lost listeners, reentrancy) applies to each view independently.

DistributedReadWriteLock(client, "/locks/catalog").use { rwLock ->
  if (rwLock.writeLock.tryLock(5.seconds)) {
    try {
      logger.info { "Got the write lock" }
    } finally {
      rwLock.writeLock.unlock()
    }
  }
}

try (DistributedReadWriteLock rwLock = new DistributedReadWriteLock(client, "/locks/catalog")) {
  if (rwLock.getWriteLock().tryLock(5, TimeUnit.SECONDS)) {
    try {
      System.out.println("Got the write lock");
    } finally {
      rwLock.getWriteLock().unlock();
    }
  }
}

Downgrade works, upgrade throws¶

Taking the read lock while holding the write lock (downgrade) is safe and supported:

DistributedReadWriteLock(client, "/locks/catalog").use { rwLock ->
  rwLock.writeLock.withLock {
    logger.info { "Writing" }

    // Downgrading write -> read is safe: take the read lock before releasing
    // the write lock, and no other writer can slip in between.
    rwLock.readLock.withLock {
      logger.info { "Still holding the write lock while reading back" }
    }
  }
}

Taking the write lock while holding the read lock (upgrade) is not:

DistributedReadWriteLock(client, "/locks/catalog").use { rwLock ->
  rwLock.readLock.withLock {
    // DON'T: upgrading read -> write self-deadlocks, because the write lock
    // waits on a predecessor this very thread is holding. The recipe throws
    // rather than hanging forever.
    // rwLock.writeLock.lock()
    logger.info { "Release the read lock first, then take the write lock" }
  }
}

DistributedSemaphore¶

At most N holders across the cluster.

// At most three clients cluster-wide run the guarded section at once.
DistributedSemaphore(client, "/semaphores/api-quota", permits = 3).use { semaphore ->
  semaphore.withPermit {
    logger.info { "Calling the rate-limited API" }
  }
}

// At most three clients cluster-wide run the guarded section at once.
try (DistributedSemaphore semaphore = new DistributedSemaphore(client, "/semaphores/api-quota", 3)) {
  semaphore.acquire();
  try {
    System.out.println("Calling the rate-limited API");
  } finally {
    semaphore.release();
  }
}

DistributedSemaphore(client, "/semaphores/api-quota", permits = 3).use { semaphore ->
  if (semaphore.tryAcquire(2.seconds)) {
    try {
      logger.info { "Got a permit; ${semaphore.availablePermits()} left (advisory)" }
    } finally {
      semaphore.release()
    }
  } else {
    logger.info { "All permits taken; shedding load" }
  }
}

try (DistributedSemaphore semaphore = new DistributedSemaphore(client, "/semaphores/api-quota", 3)) {
  if (semaphore.tryAcquire(2, TimeUnit.SECONDS)) {
    try {
      System.out.println("Got a permit; " + semaphore.availablePermits() + " left (advisory)");
    } finally {
      semaphore.release();
    }
  } else {
    System.out.println("All permits taken; shedding load");
  }
}

Semaphore holds are not lock holds¶

DistributedSemaphore deliberately does not implement EtcdLock, because its holds follow java.util.concurrent.Semaphore's rules rather than a lock's:

DistributedSemaphore(client, "/semaphores/api-quota", permits = 3).use { semaphore ->
  // Holds are instance-level, not thread-owned: unlike EtcdLock, any thread may
  // release a permit taken by another. Releases are LIFO.
  semaphore.acquire()
  semaphore.acquire()
  logger.info { "Holding two permits" }
  semaphore.release()
  semaphore.release()
}

Permit loss mirrors lock loss — a listener, connectionState → LOST, release() returning false, and an opt-in interruptOnPermitLoss:

DistributedSemaphore(client, "/semaphores/api-quota", permits = 3).use { semaphore ->
  semaphore.addPermitLostListener { cause ->
    logger.warn { "Permit lost, stop doing the guarded work: $cause" }
  }
  semaphore.withPermit { logger.info { "Working" } }
}

The permit count is fixed by the first writer¶

The canonical count is CAS-created at <semaphorePath>/permits by whichever instance reaches the path first. An instance that later names the same path with a different count throws rather than silently reconfiguring the semaphore under everyone else:

// The permit count is CAS-created at <path>/permits by whoever gets there first.
// A later instance naming the same path with a different count is a programming
// error, not a silent reconfiguration, so it throws.
try {
  DistributedSemaphore(client, "/semaphores/api-quota", permits = 5).use { semaphore ->
    semaphore.withPermit { logger.info { "Never reached if 3 was established first" } }
  }
} catch (e: SemaphorePermitMismatchException) {
  logger.error { "Requested ${e.requestedPermits}, but the path is fixed at ${e.canonicalPermits}" }
}

Coroutines¶

Every lock has suspending twins that release the thread while waiting. Note the asymmetry: withLock is scoped-only for locks (a lock is thread-owned, so the suspending version pins a confined dispatcher and releases under NonCancellable), while the semaphore also exposes split awaitAcquire/awaitRelease because its holds are instance-level. See Coroutines.

Observability¶

With the Micrometer module wired in, locks report etcd.lock.wait (tagged acquired/timeout) and etcd.lock.hold timers. Lock paths never become tags — that would blow up cardinality. See Observability.