Resilience¶

etcd is a network service, and every recipe in this library is a distributed coordination primitive layered on top of it. Networks partition, etcd restarts, leaders change. The question a recipe has to answer is not whether the connection breaks but what it does when it does.

Resilience is on by default. Every recipe ends with the same trailing parameter:

resilience: ResilienceConfig = ResilienceConfig.DEFAULT

// Every recipe ends with a `resilience: ResilienceConfig = ResilienceConfig.DEFAULT`
// parameter, so RPC retries, watch recovery, and lease healing are all on already.
PathChildrenCache(client, "/cache/orders").use { cache ->
  cache.start(true)
  logger.info { "Resilient without asking: ${cache.currentDataAsMap.size} children" }
}

// Java sees the @JvmOverloads-generated constructor without the trailing
// resilience parameter, so the DEFAULT config applies here too.
try (PathChildrenCache cache = new PathChildrenCache(client, "/cache/orders")) {
  cache.start(true);
  System.out.println("Resilient without asking: " + cache.getCurrentDataAsMap().size() + " children");
}

ResilienceConfig¶

ResilienceConfig bundles three independent policies, because the three ways a recipe talks to etcd fail differently and want different answers:

val resilience =
  ResilienceConfig(
    // Re-subscribe a dead watch fast, then back off to a 5s ceiling.
    watch = WatchResilience(RetryPolicy.exponentialBackoff(initialDelay = 100.milliseconds, maxDelay = 5.seconds)),
    // Never stop trying to heal a lease: the key it owns must come back.
    lease = LeaseResilience(RetryPolicy.forever),
    // Bound each RPC at 10s and give up after 8 attempts.
    rpc = RpcResilience(RetryPolicy.bounded(maxAttempts = 8), operationTimeout = 10.seconds),
  )

PathChildrenCache(client, "/cache/orders", resilience = resilience).use { cache ->
  cache.start(true)
}

ResilienceConfig resilience =
  new ResilienceConfig(
    new WatchResilience(RetryPolicy.exponentialBackoff()),
    new LeaseResilience(RetryPolicy.forever),
    new RpcResilience(RetryPolicy.bounded(8)));

// Java has no named or default arguments, so every preceding parameter must be
// supplied positionally: null keeps the cache's own single-threaded executor.
try (PathChildrenCache cache = new PathChildrenCache(client, "/cache/orders", null, resilience)) {
  cache.start(true);
}

Each of the three has its own DEFAULT and DISABLED constants, as does ResilienceConfig itself. ResilienceConfig.DISABLED is every sub-config disabled at once — the pre-0.12 behaviour, in which an RPC against an unreachable server parked forever, a fatally dead watcher was silently dropped, and an expired lease meant the recipe's keys were gone for good:

// The pre-0.12 behavior, should you want it back: no RPC deadline, no RPC retry,
// a fatally dead watcher stays dead, an expired lease is gone for good.
PathChildrenCache(client, "/cache/orders", resilience = ResilienceConfig.DISABLED).use { cache ->
  cache.start(true)
}

// The pre-0.12 behavior, should you want it back: no RPC deadline, no RPC retry,
// a fatally dead watcher stays dead, an expired lease is gone for good.
try (PathChildrenCache cache =
       new PathChildrenCache(client, "/cache/orders", null, ResilienceConfig.DISABLED)) {
  cache.start(true);
}

ResilienceConfig also carries the metrics sink shared by all three sub-configs. val metrics reads it; withMetrics(m) returns a copy that routes every RPC, watch, and lease funnel — plus every recipe seam — to m. See Observability.

RetryPolicy¶

A RetryPolicy answers one question: given that attempt n has failed and elapsed time has passed since the first failure, how long should we wait before trying again — or should we stop?

fun interface RetryPolicy {
  fun nextDelay(attempt: Int, elapsed: Duration): Duration?
}

Returning null means give up. That is not a detail: it is how a recovery loop terminates, and it is what turns a suspended watcher into a Failed one and a SUSPENDED connection into a LOST one.

The library ships two factories and two constants:

// Exponential backoff with jitter. Every argument has a default; this spells them
// all out, and adds a 2-minute ceiling on the total recovery window.
val paced =
  RetryPolicy.exponentialBackoff(
    initialDelay = 250.milliseconds,
    maxDelay = 15.seconds,
    factor = 2.0,
    jitterRatio = 0.25,
    maxAttempts = Int.MAX_VALUE,
    maxElapsed = 2.minutes,
  )

// A fixed delay for a fixed number of attempts — the RPC default.
val quick = RetryPolicy.bounded(maxAttempts = 4, delay = 250.milliseconds)

// The two constants: unbounded default backoff, and "never recover at all".
logger.info { "$paced / $quick / ${RetryPolicy.forever} / ${RetryPolicy.never}" }

// Only the no-argument factories are reachable from Java. Every overload that
// takes a kotlin.time.Duration compiles to a name-mangled JVM signature
// (bounded-HG0u8IE, exponentialBackoff-LRDsOJo, ...) that Java cannot name.
RetryPolicy paced = RetryPolicy.exponentialBackoff();  // 250ms -> 15s, x2, +/-25% jitter
RetryPolicy quick = RetryPolicy.bounded(4);            // 4 attempts, 500ms apart
System.out.println(paced + " / " + quick + " / " + RetryPolicy.forever + " / " + RetryPolicy.never);

exponentialBackoff jitters every delay by ±jitterRatio on purpose. A fleet of clients that all lost the same etcd will all start recovering at the same instant; without jitter they would retry in lockstep and hammer the server just as it comes back.

Because RetryPolicy is a fun interface, a lambda is a policy:

// RetryPolicy is a fun interface, so a lambda is a policy. Return null to give up.
val policy =
  RetryPolicy { attempt, elapsed ->
    when {
      // A partition this long is an incident, not a blip: stop and let the
      // Failed/Expired event escalate instead of retrying into the void.
      elapsed > 1.minutes -> null

      // Hammer briefly for an etcd restart, which is usually over in a second.
      attempt <= 3 -> 200.milliseconds

      else -> 5.seconds
    }
  }

logger.info { "First delay: ${policy.nextDelay(attempt = 1, elapsed = 0.seconds)}" }

Which failures are retried¶

RPC retries are deliberately narrow. Only three gRPC statuses are retriable:

Everything else — NOT_FOUND, INVALID_ARGUMENT, PERMISSION_DENIED, a compare failure — propagates unchanged. Those are answers, not accidents: the server understood the request and said no, and retrying would only produce the same no more slowly.

The retriable-status set lives in RpcRetry.kt and is internal — it is not part of the public surface, and there is no hook to extend it. That is intentional: the safe set is a property of etcd's semantics, not a preference.

Per-call policies¶

Every extension in common/ takes an RpcResilience as its last parameter, so a single latency-sensitive read can be tighter than the recipe holding it:

// The extension layer takes an RpcResilience per call, so a latency-sensitive read
// can opt out of retrying and use a tighter deadline than the recipe's default.
val oneShot = RpcResilience(RetryPolicy.never, operationTimeout = 2.seconds)
logger.info { "flag=${client.getValue("/config/flag", "off", oneShot)}" }

// The extension layer takes an RpcResilience per call, so a latency-sensitive
// read can opt out of retrying without changing the recipe's own config.
RpcResilience oneShot = new RpcResilience(RetryPolicy.never);
System.out.println("flag=" + KVUtils.getValue(client, "/config/flag", "off", oneShot));

Watch recovery¶

jetcd already retries transient watch-stream errors itself, transparently and with revision continuity. What it does not recover — and what WatchResilience exists for — are fatal deaths: compaction of the watched revision, halt-error statuses, and etcdserver: no leader. Before 0.12 a watcher that died this way simply stopped delivering events, silently, forever.

client.watcher(
  keyName = "/config",
  option = watchOption { isPrefix(true) },
  resilience = WatchResilience.DEFAULT,
  recoveryListener = { event -> onRecovery(event) },
  // Called only after a compaction, before re-subscribing: re-read the world,
  // rebuild whatever state was derived from the lost events, and return the
  // revision to resume from.
  resyncWith = {
    val response = client.getResponse("/config", getOption { isPrefix(true) })
    logger.info { "Resynced ${response.kvs.size} keys after compaction" }
    response.header.revision + 1
  },
) { response ->
  logger.info { "${response.events.size} event(s)" }
}.use { watcher ->
  logger.info { "Watching; closed=${watcher.isClosed}" }
}

Recovery re-subscribes from the revision just past the last event observed, so nothing is lost and nothing is duplicated — except after a compaction.

Compaction and resyncWith¶

etcd compacts old revisions away. If your watch was suspended long enough that its resume revision no longer exists, there is no way to replay the events you missed: they are gone from the server. The watcher can only re-anchor at a live revision, and whatever state you derived from the event stream is now wrong.

resyncWith is the hook that fixes it. It is invoked only after a compaction, before re-subscribing: re-read the world, rebuild your derived state from that snapshot, and return the revision to resume from (typically the GET's header.revision + 1).

progressNotify (on by default) asks etcd for periodic progress notifications so the resume revision stays fresh even on a quiet key. Without it, a watch on a key nobody writes keeps an old resume revision, and is much likelier to find it compacted away after an outage. They arrive at your watch block as WatchResponses with an empty event list.

WatchRecoveryEvent¶

fun onRecovery(event: WatchRecoveryEvent) {
  when (event) {
    // The stream errored; recovery is under way. Derived state is stale from here.
    is WatchRecoveryEvent.Suspended -> logger.warn(event.cause) { "Watch suspended on ${event.watchedKey}" }

    // Back, resuming just past the last event seen: nothing lost, nothing duplicated.
    is WatchRecoveryEvent.Resubscribed -> logger.info { "Resumed at revision ${event.resumeRevision}" }

    // Back, but etcd had compacted the resume revision away: events between
    // compactRevision and anchorRevision are gone for good.
    is WatchRecoveryEvent.Resynced -> logger.warn { "Gap from ${event.compactRevision} to ${event.anchorRevision}" }

    // The retry policy gave up. This watcher is dead and will not come back.
    is WatchRecoveryEvent.Failed -> logger.error(event.cause) { "Watch abandoned on ${event.watchedKey}" }
  }
}

Recipes that own a watch expose these through their own recovery listener (for example PathChildrenCache.addRecoveryListener) and feed them into connectionState — see Connection state.

A gap worth knowing about¶

Several withXxx { } scoped functions omit the resilience parameter even though the constructor they delegate to accepts it:

This is an inconsistency, not a design statement. Where it bites, construct the recipe directly and use Kotlin's own use { } — you lose nothing but a few characters:

val resilience = ResilienceConfig(rpc = RpcResilience(RetryPolicy.bounded(maxAttempts = 8)))

// withPathChildrenCache(...) has no `resilience` parameter, so construct the
// recipe directly and use Kotlin's own use { } instead.
PathChildrenCache(client, "/cache/orders", resilience = resilience).use { cache ->
  cache.start(true)
  logger.info { "${cache.currentDataAsMap.size} children" }
}

Java callers¶

The resilience surface is only partly reachable from Java, and it is better to know that up front than to discover it at the compiler:

The cause is the same in every case: kotlin.time.Duration is a value class, so any signature mentioning one is name-mangled on the JVM (bounded-HG0u8IE, nextDelay-dkRoYf4) into something Java cannot name or override. A Java caller can choose among the built-in policies and compose them into a ResilienceConfig, but cannot express a custom duration or a custom policy. See the Java guide.

Where to go next¶

• Leases and loss — which leases heal, which deliberately do not, and why that distinction is the most important one in this library.
• Connection state — the coarse health signal these events feed.
• Core concepts — the lifecycle and exception model every recipe shares.
• Locks — the recipes where loss is most consequential.
• Observability — metering the recovery machinery described here.