Coroutines¶

Every recipe in etcd-recipes has a blocking API, and every blocking call that waits on a network round trip has a suspending twin. The twins do not reimplement anything: they are a thin bridge over the same recipes, so a coroutine and a thread contending for the same lock are the same client to etcd.

The naming convention¶

A blocking method foo() gets a suspending twin awaitFoo(), taking the same arguments and returning the same type:

client.awaitPutValue("/config/mode", "active")
val mode = client.awaitGetValue("/config/mode", "unknown")
logger.info { "Mode is $mode" }

// Transactions get the operation timeout but are never retried: a failed commit
// is ambiguous, so the CAS retry decision stays yours.
val claimed =
  client.awaitTransaction {
    If("/leases/job-1".doesNotExist)
    Then("/leases/job-1" setTo "worker-a")
  }
logger.info { "Claimed the job: ${claimed.isSucceeded}" }

Two families deliberately break the pattern, because awaitWaitOnBarrier() and awaitDequeue() read worse than the alternative:

Everything else is await + the blocking name.

The surface, by area¶

Lifecycle — LifecycleSuspend.kt¶

Starting a recipe is itself a set of round trips, so it suspends too.

PathChildrenCache(client, "/services").use { cache ->
  // Both the start RPCs and the wait for the initial snapshot happen off-thread.
  cache.awaitStart(buildInitial = true)
  cache.awaitStartComplete(5.seconds)
  logger.info { "Cache holds ${cache.currentData.size} children" }
}

Barriers — BarrierSuspend.kt¶

DistributedDoubleBarrier(client, "/barriers/phase-1", memberCount = 5).use { barrier ->
  barrier.awaitEnter(30.seconds)
  logger.info { "All five members arrived; running phase 1" }
  barrier.awaitLeave(30.seconds)
}

See Barriers.

Counters — CounterSuspend.kt¶

DistributedAtomicLong(client, "/counters/orders").use { counter ->
  counter.awaitStart()
  val next = counter.awaitIncrement()
  val total = counter.awaitGet()
  logger.info { "Order $next of $total" }
}

Queues — QueueSuspend.kt¶

DistributedQueue(client, "/queues/jobs").use { queue ->
  queue.awaitEnqueue("job-1")

  // receive() parks the coroutine, not a thread: thousands of consumers cost
  // thousands of continuations rather than thousands of stacks.
  val item = queue.receive()
  logger.info { "Got ${item.asString}" }
}

awaitEnqueue is overloaded for ByteSequence, String, Int, and Long, matching the blocking API; the priority-queue overloads take either a UShort or an Int priority. See Queues, and Typed values for the codec-backed variants.

Work queues — WorkQueueSuspend.kt¶

Service discovery — DiscoverySuspend.kt¶

Key/value — KVSuspend.kt¶

Children — ChildrenSuspend.kt¶

Transactions — TxnSuspend.kt¶

Like its blocking twin, awaitTransaction applies the operation timeout but is never retried. A failed commit is ambiguous — it may already have been applied — so re-issuing it is the caller's decision, not the library's. See Transactions.

Leases — LeaseSuspend.kt¶

awaitLeaseRevoke is best-effort: it runs on cleanup paths where a secondary failure would mask the original problem, so failures are logged and swallowed and the TTL bounds how long the lease can outlive you. Cancellation still propagates.

Raw locks — LockSuspend.kt¶

These are pass-throughs to etcd's lock service. Prefer DistributedMutex, which manages the lease, the reentrancy, and the lock-lost notification for you.

// The raw pass-throughs; prefer DistributedMutex. awaitLock defaults to
// RpcResilience.DISABLED because the lock RPC legitimately waits server-side.
val lease = client.awaitLeaseGrant(10.seconds)
try {
  val lock = client.awaitLock("/locks/raw", lease.id, rpc = RpcResilience.DISABLED)
  try {
    logger.info { "Holding ${lock.key.asString}" }
  } finally {
    client.awaitUnlock("/locks/raw")
  }
} finally {
  client.awaitLeaseRevoke(lease)
}

Locks and semaphores — LockRecipesSuspend.kt¶

Covered in the next two sections.

Cancellation¶

This is the part worth reading properly, because it is where the coroutine layer earns its keep.

The recipe twins — everything built on a blocking recipe rather than a raw jetcd call — run their blocking body through an internal bridge (Bridges.kt) on Dispatchers.IO:

internal suspend fun <T> interruptibleOn(dispatcher: CoroutineDispatcher, block: () -> T): T
internal suspend fun <T> etcdInterruptible(block: () -> T): T   // interruptibleOn(Dispatchers.IO)

Three things follow, and they are the whole contract:

1. Cancellation interrupts the worker thread. runInterruptible means a cancelled coroutine does not merely stop waiting for the result — it interrupts the thread that is actually blocked in the recipe. Nothing is left spinning in the background.

2. The recipe's cleanup still runs, and runs to completion. The blocking recipes already handle interruption: a finally block revokes the acquisition lease, deletes the queue entry, or removes the barrier waiter. runInterruptible does not abandon the thread — it waits for that cleanup to finish before your catch sees anything. So a cancelled DistributedBarrierWithCount.await() removes its waiter and stops counting toward the barrier; a cancelled queue.receive() consumes nothing; a cancelled withLock acquisition leaves nothing queued behind a lock it will never take. Cancellation is safe, not merely fast.

3. The failure reaches you as CancellationException. The blocking RPC engine catches a mid-call InterruptedException and rethrows it wrapped in an EtcdRecipeRuntimeException — sometimes wrapped twice, because it re-sets the interrupt flag first so a subsequent RPC on the same thread can wrap it again. runInterruptible cannot recognise either as cancellation. The bridge therefore walks the entire cause chain, and since a thread inside runInterruptible is interrupted only by coroutine cancellation, an InterruptedException anywhere in that chain unambiguously means "you were cancelled" — and is re-surfaced as CancellationException.

val consumer =
  scope.launch(Dispatchers.Default) {
    DistributedQueue(client, "/queues/jobs").use { queue ->
      try {
        val item = queue.receive()
        logger.info { "Got ${item.asString}" }
      } catch (e: CancellationException) {
        // The worker thread was interrupted, the recipe's finally-block cleanup
        // (lease revoke / entry delete) already ran, and the interrupt reaches
        // you here as CancellationException — never as a raw EtcdRecipe*Exception.
        logger.info { "Consumer shut down cleanly: ${e.message}" }
        throw e
      }
    }
  }
consumer.cancel()

Composing with withTimeout / withTimeoutOrNull works exactly as you would expect, and is usually nicer than the bounded overloads when you want one deadline over several calls:

DistributedQueue(client, "/queues/jobs").use { queue ->
  // Cancelling a parked receive() consumes nothing: the wait is interrupted, the
  // recipe's cleanup runs, and the queue is left exactly as it was found.
  val item = withTimeoutOrNull(3.seconds) { queue.receive() }
  if (item == null) logger.info { "Nothing to do; idling" }
}

Locks are scoped-only; semaphores are not¶

EtcdLock gets exactly one suspending surface — withLock — while DistributedSemaphore gets a split awaitAcquire / awaitRelease pair and withPermit. That asymmetry is not an oversight; it falls straight out of what the two recipes mean.

An EtcdLock hold is owned by the acquiring thread. lock() and unlock() must run on the same thread, or unlock() throws IllegalMonitorStateException. A coroutine has no such guarantee: it can resume on a different thread of Dispatchers.IO after every suspension point. A raw suspending lock()/unlock() pair would therefore be a loaded gun — correct in testing, and broken in production the first time the dispatcher hands the continuation to another thread.

So withLock does the confining for you: it spins up one dedicated single-threaded dispatcher per call, acquires and releases on that thread, and runs your body in the caller's coroutine, where it may suspend freely.

DistributedMutex(client, "/locks/orders").use { mutex ->
  // Scoped-only: acquisition and release are confined to one dedicated thread,
  // while the body runs in the caller's coroutine and may suspend freely.
  mutex.withLock {
    logger.info { "Critical section, without pinning a thread to the wait" }
  }
}

The release leg runs under NonCancellable, so cancelling the body still releases the lock rather than leaking the hold until lease expiry. There is a bounded variant, and a null return always means "not acquired" — nothing was left queued:

DistributedMutex(client, "/locks/orders").use { mutex ->
  // A null return always means "not acquired" — nothing was left queued.
  val result =
    mutex.withLock(5.seconds) {
      logger.info { "Acquired within the timeout" }
      "done"
    }
  if (result == null) logger.info { "Someone else holds the lock; moving on" }
}

A DistributedSemaphore permit is owned by the instance, not by a thread. Any thread may release a permit that another acquired — the recipe follows java.util.concurrent.Semaphore's rules, not a lock's. Dispatcher hopping is therefore harmless, and the split surface is safe:

DistributedSemaphore(client, "/semaphores/uploads", permits = 3).use { semaphore ->
  // Permits are held by the instance, not by a thread, so the split surface is
  // safe across dispatcher hops — acquire here, release from anywhere.
  semaphore.awaitAcquire()
  try {
    val left = semaphore.awaitAvailablePermits()
    logger.info { "$left permits left" }
  } finally {
    semaphore.awaitRelease()
  }
}

withPermit is still the form to reach for, for the same reason use beats a manual close():

DistributedSemaphore(client, "/semaphores/uploads", permits = 3).use { semaphore ->
  semaphore.withPermit {
    logger.info { "At most three uploaders cluster-wide run this" }
  }
}

import io.etcd.recipes.lock.*        // ✗ brings in the blocking withLock
import io.etcd.recipes.coroutines.*  // ✗ brings in the suspending withLock

import io.etcd.recipes.lock.DistributedMutex
import io.etcd.recipes.coroutines.withLock   // ✓ unambiguous

Flows¶

Everything above is request/response. Watches, cache events, leadership hand-offs, lease transitions, and connection state are push, and they get Flow surfaces instead — see Flows.