diff --git a/AGENTS.md b/AGENTS.md
index 335ee80..5aa5bbc 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -33,8 +33,9 @@ Split what you write by *kind*, so each doc keeps exactly one job:
   history-worthy goes here: what changed and *why*, dated milestones, gotchas
   discovered, schema/protocol bugs, superseded decisions. Never rewrite or
   delete an old entry; supersede it with a new dated one.
-- **`README.md`** — only if a *user-facing* fact changed (shipped feature,
-  public API, test numbers). Most commits won't touch it.
+- **`README.md`** — **check it after every commit.** If a *user-facing* fact
+  changed (shipped feature, public API, test numbers, step/milestone status),
+  update it in the same commit so it never drifts from STATUS.md.
 
 If a commit is a legitimate pure docs catch-up (reconciling a prior
 source-only commit), bypass with `git commit --no-verify` — but that should be
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 1e8ea78..67cd98c 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -90,6 +90,7 @@ if(GTest_FOUND)
         tests/unit/ContainsKeyTest.cpp
         tests/unit/HashAwareRoutingTest.cpp
         tests/unit/ErrorHandlingTest.cpp
+        tests/unit/ServerSelectionTest.cpp
     )
 
     target_link_libraries(unit_tests
@@ -285,6 +286,30 @@ if(GTest_FOUND)
             # Note: NOT gtest_main - we provide our own main()
     )
 
+    # RetryView (Step 11b user-decided retry) integration tests (multi-server)
+    add_executable(retryview_integration_tests
+        tests/integration/RetryViewIntegrationTest.cpp
+    )
+
+    target_link_libraries(retryview_integration_tests
+        PRIVATE
+            hotrod-client
+            GTest::gtest
+            # Note: NOT gtest_main - we provide our own main()
+    )
+
+    # Ping keyless-op retry (Step 11c) integration tests (multi-server)
+    add_executable(pingretry_integration_tests
+        tests/integration/PingRetryIntegrationTest.cpp
+    )
+
+    target_link_libraries(pingretry_integration_tests
+        PRIVATE
+            hotrod-client
+            GTest::gtest
+            # Note: NOT gtest_main - we provide our own main()
+    )
+
     add_test(NAME PingIntegrationTests COMMAND ping_integration_tests)
     add_test(NAME GetIntegrationTests COMMAND get_integration_tests)
     add_test(NAME GetWithMetadataIntegrationTests COMMAND getwithmetadata_integration_tests)
@@ -300,6 +325,8 @@ if(GTest_FOUND)
     add_test(NAME ConcurrentClientsTests COMMAND concurrent_clients_tests)
     add_test(NAME HashAwareRoutingIntegrationTests COMMAND hash_aware_routing_integration_tests)
     add_test(NAME ConcurrentMultiServerTests COMMAND concurrent_multiserver_tests)
+    add_test(NAME RetryViewIntegrationTests COMMAND retryview_integration_tests)
+    add_test(NAME PingRetryIntegrationTests COMMAND pingretry_integration_tests)
 
     # Set working directory so scripts can be found
     set_tests_properties(PingIntegrationTests PROPERTIES
@@ -362,6 +389,14 @@ if(GTest_FOUND)
         WORKING_DIRECTORY ${CMAKE_BINARY_DIR}
         LABELS "integration;docker;concurrent;multi-server"
     )
+    set_tests_properties(RetryViewIntegrationTests PROPERTIES
+        WORKING_DIRECTORY ${CMAKE_BINARY_DIR}
+        LABELS "integration;docker;topology;multi-server;retry"
+    )
+    set_tests_properties(PingRetryIntegrationTests PROPERTIES
+        WORKING_DIRECTORY ${CMAKE_BINARY_DIR}
+        LABELS "integration;docker;topology;multi-server;retry"
+    )
     endif()  # NOT WIN32 (integration tests)
 
     message(STATUS "Google Test found - unit tests enabled (integration tests: non-Windows only)")
diff --git a/README.md b/README.md
index 4146820..58f547e 100644
--- a/README.md
+++ b/README.md
@@ -2,7 +2,8 @@
 
 A cross-platform C++17 implementation of the Infinispan Hot Rod protocol client.
 
-🎉 **Hash-Aware Routing Complete!** - Smart client with automatic failover.
+🎉 **Smart client** - hash-aware routing, automatic failover, full CRUD +
+metadata ops, typed error handling and user-decided retry (keyed & keyless).
 
 ## Quick Start
 
@@ -45,6 +46,10 @@ The quickstart demonstrates:
 - REMOVE operations (deleting entries)
 - Clean disconnect
 
+A second example, `retry.cpp`, demonstrates the user-decided retry loop
+(`cache.excluding(e)` on a caught `HotRodClientException`) for both keyed
+operations and keyless `ping`.
+
 See [`examples/quickstart/README.md`](examples/quickstart/README.md) for full documentation.
 
 ## Building
@@ -109,7 +114,7 @@ ctest -C Release --output-on-failure
 
 ## Testing
 
-### Unit Tests (155 tests)
+### Unit Tests (204 tests)
 ```bash
 cd build
 ./unit_tests
@@ -118,18 +123,20 @@ cd build
 ctest -R UnitTests --output-on-failure
 ```
 
-### Integration Tests (38 tests)
+### Integration Tests (84 tests, 17 suites)
 Requires Docker to run Infinispan server:
 ```bash
 cd build
 
-# Run all integration tests (starts server automatically)
+# Run all integration tests (starts server automatically) — representative subset:
 ./ping_integration_tests
-./get_integration_tests  
+./get_integration_tests
 ./put_integration_tests
 ./remove_integration_tests
-./topology_integration_tests         # Multi-node cluster tests
+./topology_integration_tests            # Multi-node cluster tests
 ./hash_aware_routing_integration_tests  # Smart routing tests
+./retryview_integration_tests           # User-decided retry (keyed)
+./pingretry_integration_tests           # User-decided retry (keyless ping)
 
 # Or run all via CTest
 ctest --output-on-failure
@@ -143,18 +150,20 @@ ctest --output-on-failure
 > Design rationale is in [docs/DECISIONS.md](docs/DECISIONS.md).
 
 **Current Progress**: Full CRUD (PING/GET/PUT/REMOVE) + SCRAM auth + topology
-awareness + hash-aware routing + async multiplexing — all shipped. **Next:**
-metadata / version-based operations (Step 10). See [docs/STATUS.md](docs/STATUS.md)
-for live status. Per-step milestone history (frozen at Step 9) is archived at
+awareness + hash-aware routing + async multiplexing + metadata/version-based
+operations (Step 10) + typed error handling and user-decided retry, keyed and
+keyless (Step 11) — all shipped. **Next:** bulk operations (Step 12) or
+multiplexing benchmarks. See [docs/STATUS.md](docs/STATUS.md) for live status.
+Per-step milestone history (frozen at Step 9) is archived at
 [docs/archive/PROGRESS.md](docs/archive/PROGRESS.md).
 
 _(Step numbers follow [`../hotrod-foundry/ROADMAP.md`](../hotrod-foundry/ROADMAP.md);
 [`docs/STATUS.md`](docs/STATUS.md) is authoritative for which steps are done — not
 any headline count.)_
 
-**Test Results** (verified 2026-09-16):
-- Unit Tests: **145/145 passing** ✅
-- Integration Tests: **46/46 passing** ✅ (against live Infinispan via Docker)
+**Test Results** (verified 2026-09-25):
+- Unit Tests: **204/204 passing** ✅
+- Integration Tests: **84/84 passing** ✅ across 17 suites (against live Infinispan via Docker)
 
 ## Features
 
@@ -174,6 +183,12 @@ any headline count.)_
 - ✅ **GET operation** (read from cache with failover)
 - ✅ **PUT operation** (write to cache with lifespan/maxIdle)
 - ✅ **REMOVE operation** (delete from cache)
+- ✅ **Metadata / version-based operations** (`getWithMetadata`, `putIfAbsent`,
+  `replace`, `containsKey`, `removeWithVersion`, `replaceWithVersion`) (Step 10)
+- ✅ **Typed error handling** (`HotRodClientException` with phase / server status /
+  tried nodes) + `isTransient` / `outcomeUncertain` classification (Step 11a)
+- ✅ **User-decided retry** (`cache.excluding(e)` bound view) for keyed ops and
+  keyless `ping`, with automatic before-send failover (Step 11b/11c)
 - ✅ **Integration test framework** (GoogleTest + Docker + multi-node clusters)
 
 ### Full CRUD with Smart Routing and Async Operations
@@ -217,10 +232,11 @@ for (int i = 0; i < 1000; i++) {
 > the 1–3 immediate items in
 > [⏭ Next steps](docs/STATUS.md#-next-steps-start-here). This is just a summary.
 
-- **Step 10 — Metadata / version-based operations** (next): `getWithMetadata`,
-  `replaceWithVersion`, `removeWithVersion`, `putIfAbsent`, `containsKey`
-- **Step 11 — Error handling**: ERROR response parsing, exception hierarchy, retries
-- **Step 12 — Bulk operations**: `GET_ALL`, `PUT_ALL`, `BULK_GET`
+- ✅ **Step 10 — Metadata / version-based operations** (shipped): `getWithMetadata`,
+  `replaceWithVersion`, `removeWithVersion`, `putIfAbsent`, `replace`, `containsKey`
+- ✅ **Step 11 — Error handling** (shipped): ERROR response parsing, typed exception,
+  user-decided retry (keyed + keyless)
+- **Step 12 — Bulk operations** (next): `GET_ALL`, `PUT_ALL`, `BULK_GET`
 - Also: multiplexing benchmarks, TLS/SSL support
 
 ## Project Structure
@@ -246,8 +262,8 @@ cpp17-client/
 │   ├── hash/                    # MurmurHash3 + consistent hashing
 │   └── operations/              # Hot Rod operations (with failover)
 ├── tests/                       # Test suite
-│   ├── unit/                    # 155 unit tests
-│   └── integration/             # 38 integration tests (multi-node clusters)
+│   ├── unit/                    # 204 unit tests
+│   └── integration/             # 84 integration tests, 17 suites (multi-node clusters)
 ├── examples/                    # Usage examples
 │   └── quickstart/              # Simple GET/PUT/REMOVE example
 ├── scripts/                     # Test infrastructure
@@ -317,10 +333,14 @@ In short, following the hotrod-foundry porting guidelines:
 - ✅ **v0.5.0** (2026-06-12): REMOVE operation complete (Step 9) - **Full CRUD!**
 - ✅ **v0.6.0** (2026-06-17): Topology awareness (Step 4)
 - ✅ **v0.7.0** (2026-06-17): Hash-aware routing with failover - **Smart Client!**
-- 🎯 **v0.8.0** (upcoming): Metadata operations (Step 10)
+- ✅ **v0.8.0** (2026-09-20): Metadata / version-based operations (Step 10)
+- ✅ **v0.9.0** (2026-09-25): Typed error handling + user-decided retry, keyed &
+  keyless (Step 11)
+- 🎯 **v0.10.0** (upcoming): Bulk operations (Step 12)
 
 ---
 
 **Maintained by**: rigazilla  
 **Status**: Active development  
-**Current Step**: Hash-aware routing complete! Smart client with automatic failover. ✅
+**Current Step**: Error handling + user-decided retry complete (Step 11, keyed &
+keyless). Next: bulk operations (Step 12). ✅
diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md
index f3dc915..0210995 100644
--- a/docs/DECISIONS.md
+++ b/docs/DECISIONS.md
@@ -229,3 +229,388 @@ retry user-decided rather than automatic (2026-09-21 entry).
 the two axes and forces the library to pre-judge user idempotency); follow Java
 and treat 0x86 as non-retriable (rejected — the ambiguity is already carried by
 `outcomeUncertain()`, so the extra futility signal is safe and useful).
+
+---
+
+## 2026-09-25 — Step 11b (slice 1): exclusion-aware server selection
+
+**What shipped:** the first slice of Step 11b — the routing mechanism a
+user-driven retry needs, with no behavioural change to existing (non-retrying)
+call sites.
+
+- **`orderKeyCandidates()`** (new `include/hotrod/ServerSelection.h`, header-only,
+  pure): given owners, all topology servers, and an exclusion set, returns the
+  ordered candidate list — owners first (in owner order), then non-owner proxy
+  fallback — with excluded nodes and duplicates removed. Pulled out precisely
+  because the old inline selection logic was untestable (needs live sockets);
+  the pure decision now has 7 unit tests (`ServerSelectionTest.cpp`).
+- **`RetryContext`** (new `include/hotrod/RetryContext.h`): pure-value
+  user→library input, currently just `excludeNodes`. Deliberately does NOT carry
+  `proxyToNonOwner` — that is client-level routing policy (D5), landing in
+  slice 2.
+- **`selectServerForKey`** rewritten over `orderKeyCandidates`; gains defaulted
+  `RetryContext` + optional `triedOut` out-param that reports every node it
+  attempted (for accumulating `triedNodes` across retries in slice 3). On
+  exhaustion it now computes `ownersExhausted` honestly (an owner counts as
+  exhausted if tried this attempt *or* in `excludeNodes` from a prior one)
+  instead of the previous hard-coded `true`.
+
+**Why sliced this way:** exclusion-aware selection is the foundation both the
+op-level `RetryContext` threading (slice 3) and `proxyToNonOwner` (slice 2)
+build on, and it is the one piece testable without a cluster. Keeping it a pure
+function keeps the I/O (connection acquisition, which can throw per candidate)
+separate from the decision.
+
+**Note:** the design-doc example loop still referenced a removed `e.retriable`
+field; corrected to `isTransient(e)` while here.
+
+---
+
+## 2026-09-25 — Step 11b (slice 2): proxyToNonOwner client flag
+
+**What shipped:** client-level `proxyToNonOwner` (default `true`, matching Java),
+gating whether hash-aware selection falls through from a key`s owners to a
+non-owner that proxies to the real owner.
+
+- `setProxyToNonOwner()` / `getProxyToNonOwner()` on `RemoteCache`; backing
+  `proxyToNonOwner_` member defaulted `true`.
+- The gate lives in the pure `orderKeyCandidates()` (new `bool` param): when
+  false it stops after owners, so `selectServerForKey` exhausting the owners
+  throws the transient BeforeSend exception with `ownersExhausted=true` instead
+  of silently proxying. 3 new unit tests (10 total in `ServerSelectionTest`).
+
+**Why a client-level flag, not `RetryContext` (D5):** proxy fallback is routing
+policy, not per-call input — it mirrors how Java models balancing/maxRetries at
+the client, and keeps `RetryContext` a pure exclusion carrier. **Scope:** the
+flag governs only the BeforeSend connection-fallback inside `selectServerForKey`;
+it does NOT make AfterSend retry automatic (that stays user-decided, D2). So
+`true` means "Java-like routing fallback", not "fully Java-automatic".
+
+Unit: 197/197 (was 194).
+
+## 2026-09-25 — Step 11b (slice 3): op-level retry via a bound view, NOT a per-op parameter
+
+**What shipped:** the retry path that lets a caught exception feed the next
+attempt, without putting any retry machinery on the base operation signatures.
+
+- New `RetryView` (`RetryView.h`): a lightweight, non-owning view holding
+  `RemoteCache* + RetryContext`. It exposes the same key-routed ops, each
+  forwarding to a private `RemoteCache::…Impl(args, const RetryContext&)`.
+- `RemoteCache::excluding(std::vector<ServerAddress>)` and
+  `excluding(const HotRodClientException&)` return a `RetryView`; the second
+  seeds the exclusion set directly from a caught exception's `triedNodes`.
+- Each of the 9 ops split into a thin public forwarder (`return getImpl(key,
+  RetryContext{});`) + a private `…Impl` carrying the body. The public
+  signatures are **byte-for-byte unchanged**.
+- `triedNodes` accumulation: every attempt (including the plain first call)
+  unions the nodes it touched into any thrown `HotRodClientException` via the new
+  pure `unionNodes()` helper (`ServerSelection.h`), folding
+  `excludeNodes ∪ tried-this-attempt ∪ e.triedNodes`. A file-private
+  `rethrowWithTriedNodes()` in `RemoteCache.cpp` does this on the async path;
+  the selection-exhaustion and inline-status throws do it inline. 4 new unit
+  tests (14 total in `ServerSelectionTest`).
+
+**Why a bound view instead of a defaulted `RetryContext` on each op (supersedes
+the fork-1.b sketch in the 2026-09-21 entry):** the original 11b plan added an
+optional `RetryContext` parameter to every op. The user rejected that — the
+requirement is that the base call (`cache.get(key)`) stay completely free of
+retry machinery in its signature, while it's fine for the *exception* to carry
+tried-node data. The bound view satisfies both: base ops are pristine and
+retry-free; opting into exclusion is an explicit, separate `cache.excluding(e)`
+step on the retry path; and `triedNodes` is populated on every exception
+(first call included) so a retry loop always has the exclusion set it needs.
+This supersedes only the *mechanism* (per-op parameter → bound view); the fork
+1.b rationale (retry threads exclusion through selection; policy/idempotency stay
+the user's call, D2/D3) is unchanged.
+
+Unit: 201/201 (was 197). Full multi-target build clean under `-Werror`.
+
+## 2026-09-25 — Two intentional divergences from the Java client's routing/retry
+
+Cross-checked our hash-aware selection and retry model against the authoritative
+Java client (verified in-source, not from prose):
+`SegmentConsistentHash.getServer` and the `RoundRobinBalancingStrategy` /
+`OperationDispatcher` retry path. Two deliberate divergences, chosen with eyes
+open and kept:
+
+**1. Candidate ordering — backup-first + strict exclusion (routing).**
+Java, hash-aware with the default RoundRobin balancer:
+- `SegmentConsistentHash.getServer()` returns **`segmentOwners[segmentId][0]` —
+  the primary owner only.** Backups get no priority.
+- On failure it falls to `RoundRobinBalancingStrategy.nextServer(failedServers)`,
+  round-robining over the **entire topology** (shared atomic index), skipping the
+  per-op failed set **best-effort** (it can return a failed server once it has
+  looped `failedServers.size()` times).
+
+Ours (`orderKeyCandidates`, `proxyToNonOwner=true`): candidates are
+`[primary, backups…, then all other servers]` in deterministic owner order, and
+excluded nodes are **strictly** removed (never returned). **Why kept (option 1):**
+preferring backup owners avoids a server-side proxy hop (backups hold the data),
+the order is deterministic (easier to reason about/test), and strict exclusion is
+what makes the user-driven retry loop actually converge instead of possibly
+re-hitting a dead node. Trade-off: not byte-for-byte Java behavior. This is the
+routing analogue of the 0x86 divergence already recorded (2026-09-25, 11a).
+
+**2. After-send retry is delegated to the user (not automatic).**
+Java's balancer + `maxRetries` re-dispatches **after-send** failures across
+servers automatically. For non-idempotent ops with an **uncertain outcome**
+(`COMMAND_TIMEOUT` 0x86 / `FailurePhase::AfterSend` — the server may already have
+applied the op) that risks double-apply or clobbering a concurrent write. We do
+NOT auto-retry: we surface `phase` + `outcomeUncertain()` and let the caller —
+who alone knows whether *this* op is replay-safe — decide. **Why kept:** strictly
+more correct for the non-idempotent case; the idempotent read case costs only a
+three-line retry loop. Trade-off: naive callers get no automatic retry. This
+reinforces D2 (retry user-decided) and D4 (classify futility/ambiguity, don't
+auto-decide safety); it is a considered improvement over Java's default, not an
+omission.
+
+## 2026-09-25 — Step 11b (slice 4): thread-safety for the routing/pool state
+
+**What shipped:** the selection path can now be driven from user/retry threads
+concurrently with the read-loop thread's topology callbacks, which is exactly the
+concurrency the RetryView retry loop introduces.
+
+- **One `stateMutex_`** (mutable) guards the three shared members —
+  `topology_`, `consistentHash_`, `connectionPool_`. It is acquired at the top of
+  the two entry points that touch them: `handleTopologyUpdate` (read-loop thread,
+  writes) and `selectServerForKey` (user/retry threads, reads + pool mutate).
+  Their internal helpers — `cleanupStaleConnections` and `getConnectionForServer`
+  — assume the lock is held and do **not** re-lock (documented; avoids a
+  recursive mutex).
+- **Connection lifetime via `shared_ptr`.** The pool and the default connection
+  are now `shared_ptr<MultiplexedConnection>`; `selectServerForKey` returns a
+  `shared_ptr` and each op captures it into its response lambda. This closes a
+  use-after-free the concurrent path would otherwise expose: previously
+  `selectServerForKey` handed back a raw pointer that `cleanupStaleConnections`
+  could `erase`/destroy while the op was mid-`execute()`. Now a connection handed
+  to an in-flight op stays alive for the op's duration even if a topology update
+  evicts it from the pool.
+
+**Deadlock / lock-ordering analysis (verified in-source):**
+- `connect()` never fires the topology callback synchronously — `onTopologyUpdate_`
+  is invoked only from `readLoop()` on the connection's own read thread — so
+  holding `stateMutex_` across `getConnectionForServer`→`connect()` cannot
+  self-deadlock.
+- In `readLoop`, `onTopologyUpdate_` is called **after** `pendingMutex_` is
+  released, so the read thread holds no MultiplexedConnection-internal lock when
+  `handleTopologyUpdate` acquires `stateMutex_`. There is thus no cycle with
+  `selectServerForKey` (which takes `stateMutex_` first, then connection-internal
+  locks via `connect()`).
+
+**Trade-offs / known limits (deliberate, noted for the next session):**
+- `stateMutex_` is held across the blocking `connect()` inside
+  `getConnectionForServer`, so a slow connect serializes other selections and can
+  briefly delay a topology callback. Correctness-first; a double-checked
+  connect-outside-lock is possible later if it ever matters.
+- The debug/test accessors `getTopology()` / `getConsistentHash()` still return
+  references to the guarded state and are **not** safe to read concurrently with
+  live ops / topology churn (documented on them); `getTopologyId()` is locked and
+  safe. Converting the reference getters to by-value was out of scope (pervasive
+  test use, chaining, ConsistentHash copyability).
+- Teardown still assumes single-threaded shutdown (pool connections' read threads
+  stop when `connectionPool_` is destroyed); not a concurrent-operation concern,
+  left as-is.
+
+Validation: full multi-target build clean under Release `-Werror`; unit 201/201.
+Thread-safety itself is best validated by the existing concurrent integration
+suites (`ConcurrentClientsTest`, `ConcurrentMultiServerTest`) and a TSan run —
+these need live servers/Docker and were not run in this session.
+
+## 2026-09-25 — Step 11b (slice 4): TSan validation + test-harness race fix
+
+**TSan run (done this session, against live Docker cluster).** The concurrent
+integration suites were run both functionally (8/8 pass, including
+`ConcurrentWithFailover`, which kills node2 mid-run) and under ThreadSanitizer
+(`-fsanitize=thread`, dedicated `build-tsan/`).
+
+**Result: slice 4's synchronization held clean.** TSan reported **zero** races in
+production code — no frame in any race stack touches `RemoteCache.cpp` or
+`MultiplexedConnection.cpp` (`selectServerForKey`, `handleTopologyUpdate`,
+`getConnectionForServer`, `connectionPool_` all absent). The `stateMutex_` +
+`shared_ptr` design (previous entry) is confirmed race-free under the retry
+loop's concurrency.
+
+**One test-harness race found and fixed.** All 9 TSan warnings pointed at the
+same line — `TopologyTestFixture.h` `createClient()` doing an unsynchronized
+`clients.push_back()` on a shared `std::vector<unique_ptr<RemoteCache>>` while
+`ConcurrentMultiServerTest.MultipleClientsInterlaced` calls it from several
+threads. `std::vector::push_back` is not safe for concurrent callers (a
+reallocation moves the buffer out from under another thread's write). This is
+test scaffolding, predates slice 4, and never corrupted production state — but a
+dirty TSan baseline hides *future* real races. **Fix:** a `clientsMutex` guards
+the `push_back`; `TearDown` stays lockless (runs single-threaded after joins).
+Test-only change; behavior unchanged; concurrent suites still build clean.
+
+## 2026-09-25 — Step 11b (slice 5): RetryView end-to-end integration test
+
+**What shipped:** `tests/integration/RetryViewIntegrationTest.cpp` (3 tests,
+verified 3/3 against a live 3-node Docker cluster), wired into CMake/ctest as
+`RetryViewIntegrationTests`. It is deliberately distinct from the existing
+failover tests (`ConcurrentWithFailover`, `FailoverToBackupOwner`), which cover
+*automatic* before-send re-routing; this suite asserts the *explicit*
+user-driven `excluding()` dispatch that is the point of slice 3.
+
+- **`ExcludingRoutesAroundOwner`** (deterministic, no kill): `cache.excluding(
+  {primaryOwner}).get(key)` still returns the value, proving the RetryView routes
+  the request to a different node (backup owner or non-owner proxy). Isolates the
+  routing decision from failure timing.
+- **`ProxyDisabledOwnersExhaustedThrows`** (deterministic, no kill): with
+  `setProxyToNonOwner(false)` and every owner excluded, selection has no
+  candidate and throws `HotRodClientException{phase=BeforeSend,
+  ownersExhausted=true}`, `isTransient(e)==true`, with the excluded owners
+  present in `triedNodes`. Locks in the D5 contract (refuse to silently proxy;
+  hand the decision back).
+- **`UserRetryLoopAfterPrimaryKilled`** (end-to-end, kills the primary owner):
+  runs the documented catch→`excluding(e)` loop from ERROR_HANDLING_DESIGN §3.2.
+
+**Design note — why the loop test tolerates two recovery paths.** After the
+primary owner is killed, a plain `get` can recover *either* by throwing (dead
+connection / owner unreachable) and being retried via `excluding(e)`, *or* by
+automatic before-send re-routing once the topology update lands — which path wins
+is a timing race we cannot make deterministic without artificial fault injection.
+The test therefore asserts the invariant that actually matters (the value is
+recovered) and, *only if* an exception was caught, that the failure was transient,
+that each attempt widened `triedNodes`, and that the killed primary ended up in
+the exclusion set. The first observed run recovered via automatic re-routing;
+the deterministic `excluding()` dispatch is nonetheless fully covered by the two
+no-kill tests above, so RetryView routing does not depend on that race. Trade-off
+accepted: no artificial after-send fault injection (would need a fault hook in
+the transport) — deferred as not worth the complexity for this step.
+
+## 2026-09-25 — Step 11b (slice 6): retry-loop example — **Step 11b complete**
+
+**What shipped:** `examples/quickstart/retry.cpp` (built as a second target
+alongside `quickstart`) plus a README section. Two reusable loops —
+`getWithRetry` (idempotent) and `putWithRetry` (shows the extra
+`outcomeUncertain()` gate a non-idempotent write needs) — implement the
+documented pattern: retry-free first call, then `cache.excluding(e)` seeded from
+the caught exception's `triedNodes`, stopping on `!isTransient(e)`. Builds
+warning-clean under `-Wall -Wextra -Werror -pedantic`; runs end-to-end against a
+live single server (the retry branch only fires on a cluster with a node killed
+mid-run, as the file/README note).
+
+**The clarification folded in (user's Q from the prior session).** The example
+header and README state explicitly that `proxyToNonOwner=true` does **not** let a
+caller skip the catch block: the flag only widens the *connection-selection*
+candidate pool for a single pre-send dispatch, but one operation still executes
+on exactly one node, so any after-send/server-error failure still surfaces as an
+exception the caller must catch and retry. This closes the open question from the
+2026-09-25 "two intentional divergences" discussion and completes Step 11b.
+
+**Deliberately left out of the example:** artificial fault injection to force the
+retry branch. Keeping the example runnable against a plain single server (and
+pointing at `RetryViewIntegrationTest` for the kill-a-node scenario) was judged
+clearer than shipping an example that needs a cluster to run at all.
+
+## 2026-09-25 — Step 11c (slice 1): keyless-op retry — ping mirrors the keyed strategy
+
+**Decision.** Keyless operations get the *same* two-tier retry model as keyed ops
+(see 2026-09-21 / D2): (1) **automatic before-send failover** — when the chosen
+server is unreachable the client silently sweeps to the next server in topology
+order; (2) **user-decided after-send retry** — once a request has been sent, any
+AfterSend/ServerError failure surfaces as a `HotRodClientException` carrying
+`triedNodes`, and the caller retries via `cache.excluding(e).ping()`. Replay
+safety stays the caller's call. `ping()` is the first keyless op converted; the
+implementation is generic (`selectAnyServer`) so future non-key ops (server
+stats/admin) reuse it — satisfying the user's directive that this be *"the
+approach for all the non-key operations."*
+
+**Mechanism.** `ping()` now delegates to `pingImpl(const RetryContext&)` (the same
+base-call / `…Impl(ctx)` split every keyed op uses), and `RetryView::ping()`
+forwards the bound exclusion set. Routing goes through a new
+`selectAnyServer(ctx, triedOut)`, which reuses the pure, unit-tested
+`orderKeyCandidates({}, allServers, ctx.excludeNodes)` — an **empty owner list**
+means "every server is a candidate, in topology order, minus the exclusion set,"
+so no new pure helper was needed. It sweeps candidates, returning the first
+reachable connection (recording tried nodes via the out-param), and on total
+exhaustion throws `HotRodClientException{BeforeSend, ownersExhausted=false, …}`.
+
+**Two deliberate choices.**
+- **`ownersExhausted` is always `false` for keyless ops** — there is no "owner"
+  concept for a keyless request, so the flag that means "proxy was refused" is
+  meaningless here; the exhaustion is a plain BeforeSend "no servers reachable."
+- **Early-ping seed fallback (per user note).** Ping is typically one of the very
+  first operations, before any topology update has arrived, so
+  `selectAnyServer` returns the seed `connection_` when the topology is empty
+  (rather than throwing "no servers"). This preserves today's first-ping behavior
+  exactly (verified: `PingIntegrationTest` still 5/5).
+
+**Order divergence from Java (consistent with our keyed determinism).** Java's
+user-facing cache ping (`CachePingOperation`, `supportRetry=true`) routes via the
+keyless *round-robin* balancer and auto-retries across servers excluding a
+`failedServers` set up to `maxRetries`. We instead use **deterministic topology
+order** (matching our keyed-routing determinism divergence) and keep the
+after-send retry **user-decided** rather than automatic (D2). Java's other ping,
+`NoCachePingOperation` (`supportRetry=false`, per-connection health/liveness,
+force-written to one channel), has no user-facing analog here and is unaffected.
+
+**Tests.** 3 keyless-ordering unit tests added to `ServerSelectionTest`
+(`KeylessNoOwnersYieldsAllServersInOrder`, `KeylessExcludesTriedNodes`,
+`KeylessAllExcludedYieldsEmpty`) — 201→204, all green; `PingIntegrationTest` still
+5/5 with the new routing. Integration coverage for the kill-and-recover path is
+slice 2.
+
+## 2026-09-25 — Step 11c (slice 2): keyless-ping retry — end-to-end integration test
+
+**What shipped:** `tests/integration/PingRetryIntegrationTest.cpp` (3 tests,
+verified 3/3 against a live 3-node Docker cluster), wired into CMake/ctest as
+`PingRetryIntegrationTests`. It is the keyless mirror of
+`RetryViewIntegrationTest`, asserting that `ping()`'s `selectAnyServer` path
+behaves like the keyed retry path on real infrastructure:
+
+- **`ExcludingRoutesToAnotherServer`** (deterministic, no kill): a plain `ping()`
+  succeeds, and `cache.excluding({firstCandidate}).ping()` also succeeds — the
+  keyless op is routed to a different server.
+- **`AllServersExcludedThrowsBeforeSend`** (deterministic, no kill): excluding
+  every topology server leaves `selectAnyServer` no candidate and throws
+  `HotRodClientException{phase=BeforeSend}`, `isTransient==true`, with every
+  excluded server in `triedNodes`. Crucially it asserts **`ownersExhausted==false`**
+  — the one behavioural difference from the keyed
+  `ProxyDisabledOwnersExhaustedThrows` case (a keyless op has no owners).
+- **`RecoversAfterFirstCandidateKilled`** (kills a node): kills the server the
+  keyless ping targets first, then drives the documented catch→`excluding(e).ping()`
+  loop; asserts recovery on a surviving node, and — only if an exception was
+  caught — that the failure was transient and the killed node ended up in the
+  exclusion set.
+
+**Two integration-harness notes worth recording.**
+- **Topology reports internal container addresses.** On the Linux Docker bridge
+  the Hot Rod topology advertises container IPs (e.g. `172.18.0.2:11222`), which
+  the client reaches directly — *not* the published `localhost:1132x` ports the
+  test harness tracks. So a topology address cannot be mapped to a node number by
+  matching `MultiServerTestEnvironment` ports (returns -1). The test maps by
+  **topology position** instead (topology is ordered node1, node2, …), the same
+  convention `RetryViewIntegrationTest` already relies on.
+- **First observed recovery was via automatic before-send re-routing, not the
+  catch path.** Killing the first candidate node made `selectAnyServer`'s
+  connect to that node fail, so it swept to the next server *before sending* and
+  the very first `ping()` succeeded (no exception). The test tolerates both this
+  path and the throw-then-`excluding(e)` path, exactly as the keyed loop test
+  does — which path wins is a timing race we don't force with artificial fault
+  injection. The deterministic `excluding()` dispatch and the exhaustion throw
+  are fully covered by the two no-kill tests, so keyless routing does not depend
+  on that race.
+
+## 2026-09-25 — Step 11c (slice 3): keyless-ping retry example — **Step 11c complete**
+
+**What shipped:** `examples/quickstart/retry.cpp` gained a keyless `pingWithRetry`
+loop (built into the existing `retry` target), plus doc updates to
+`examples/quickstart/README.md` and `docs/ERROR_HANDLING_DESIGN.md` (a new Step 11c
+checklist block under §5). The loop is byte-for-byte the same shape as
+`getWithRetry` — retry-free first call, then `cache.excluding(e).ping()` seeded
+from the caught exception — underscoring that keyless ops need no special handling.
+`main()` calls `pingWithRetry` right after `connect()`, the realistic spot: ping is
+usually the first op, before any topology has arrived (the seed-connection fallback
+covers that). Compiles warning-clean under `-Wall -Wextra -Werror -pedantic` and
+runs end-to-end against a live single server (the retry branch only fires on a
+cluster with a node killed mid-run, as the file/README note; the kill scenario is
+covered by `PingRetryIntegrationTest`).
+
+**Doc note folded in.** The README/example spell out the one visible difference of
+the keyless path — a keyless exhaustion reports `ownersExhausted == false` (no
+owners) — so a reader porting the keyed loop to `ping` isn't surprised.
+
+With slices 1–3 done, **Step 11c is complete**: keyless ops mirror the keyed
+two-tier retry model (auto before-send failover, user-decided after-send), the
+implementation (`selectAnyServer`) is generic for future non-key ops, and the
+behaviour is unit-, integration-, and example-covered.
diff --git a/docs/ERROR_HANDLING_DESIGN.md b/docs/ERROR_HANDLING_DESIGN.md
index 41a5102..2c8339b 100644
--- a/docs/ERROR_HANDLING_DESIGN.md
+++ b/docs/ERROR_HANDLING_DESIGN.md
@@ -6,8 +6,8 @@
 > keep the one-line pointer in [`STATUS.md`](STATUS.md) → *Next steps* current.
 >
 > **Created:** 2026-09-21 · **Refined:** 2026-09-25 (D4 classification, D5
-> proxy-to-non-owner, Java cross-check) · **Status:** design agreed,
-> implementation not started
+> proxy-to-non-owner, Java cross-check) · **Status:** 11a shipped; 11b slices 1–2
+> (exclusion-aware selection, proxyToNonOwner) shipped 2026-09-25; slices 3–4 remain
 
 ---
 
@@ -115,6 +115,21 @@ connection-fallback inside `selectServerForKey`. Java also auto-re-dispatches
 transient exception with `ownersExhausted = true` so the user drives the proxy
 step.
 
+**Two intentional divergences from Java, even with the flag `true`** (verified
+in-source 2026-09-25; full rationale in DECISIONS.md, 2026-09-25 "Two intentional
+divergences" entry — kept as *option 1*):
+1. **Candidate order.** Java routes to the **primary owner only**
+   (`SegmentConsistentHash.getServer` → `segmentOwners[seg][0]`), then round-robins
+   over the whole cluster (backups get no priority) with *best-effort* exclusion.
+   Ours orders `[primary, backups…, then non-owners]` deterministically with
+   **strict** exclusion. Preferring backups avoids a proxy hop; strict exclusion
+   makes the user retry loop converge.
+2. **After-send retry.** Java auto-retries after-send failures across servers up to
+   `maxRetries`. We do **not** — the caller decides (D2/D4), since only the caller
+   knows if a non-idempotent op is replay-safe. So one call's connection sweep
+   covers reachability across the cluster, but a node that *connects then errors*
+   is surfaced to the user, not silently re-dispatched.
+
 ---
 
 ## 3. The design (target)
@@ -156,9 +171,10 @@ exactly what lets the user make the idempotency call:
 completed). `triedNodes` accumulates across successive retries (each produced
 error carries the *union*).
 
-### 3.2 Retry via the cache (11b)
-Ops gain an optional retry/exclusion context, defaulted so existing call sites are
-untouched:
+### 3.2 Retry via the cache — a bound view (11b)
+The retry/exclusion context is **not** a parameter on the base operations. The
+base signatures stay completely retry-free; a caller opts into exclusion by
+asking the cache for a *bound view* that carries a `RetryContext`:
 
 ```cpp
 struct RetryContext {                          // USER→library input only
@@ -168,26 +184,37 @@ struct RetryContext {                          // USER→library input only
     //     (config), matching how Java models balancing/maxRetries (D5).
 };
 
-// existing signature keeps working:
+// base op — signature unchanged, zero retry machinery:
 std::future<std::optional<ByteArray>> get(const ByteArray& key);
-// retrying caller passes context built from the caught exception:
-std::future<std::optional<ByteArray>> get(const ByteArray& key,
-                                          const RetryContext& ctx);
+
+// bound view: same ops, each dispatched avoiding the excluded nodes:
+RetryView RemoteCache::excluding(std::vector<ServerAddress> excludeNodes);
+RetryView RemoteCache::excluding(const HotRodClientException& e);  // seed from e.triedNodes
 ```
 
+`RetryView` is a lightweight, non-owning view (`RemoteCache* + RetryContext`)
+whose ops forward to private `RemoteCache::…Impl(args, const RetryContext&)`
+methods. **Design change (supersedes the earlier per-op `RetryContext`
+parameter):** the base call must stay free of retry machinery in its signature;
+it is fine for the *exception* to carry tried-node data. See the 2026-09-25
+slice-3 entry in DECISIONS.md.
+
 Typical user loop:
 ```cpp
-RetryContext ctx;
 for (;;) {
-    try { return cache.get(key, ctx).get(); }
+    try { return cache.get(key).get(); }        // first call: pristine, retry-free
     catch (const HotRodClientException& e) {
-        if (!e.retriable) throw;
+        if (!isTransient(e)) throw;             // futility is a free function, not a field
         // user decides based on e.phase whether THIS op is safe to retry
-        ctx.excludeNodes = e.triedNodes;
+        return cache.excluding(e).get(key).get();  // avoids e.triedNodes
     }
 }
 ```
 
+Even the pristine first call populates `e.triedNodes`, so `cache.excluding(e)`
+always has the exclusion set it needs — accumulated monotonically across
+attempts via the pure `unionNodes()` helper.
+
 `selectServerForKey` gains an **exclusion-aware** variant so a retry avoids the
 dead node(s). Retry is then just the normal op path with an exclusion set — **no
 closure capture, no shared dispatcher, no templated exception.**
@@ -279,7 +306,7 @@ What each fork adds **on top of B**:
 | Exception shape | Templated `RetriableError<T>` + non-template base so `retry()` can return the op's value (M) | Pure data, one type (0) |
 | Re-dispatch machinery | Replayable closure captured into the error; reify each op into a re-invocable form (M) | Normal op path + exclusion arg — no capture (0) |
 | Lifetime | Extract a shared dispatcher/session so the error can re-enter after crossing the `future` boundary; permanent "stashed error outlives client → UB" hazard (L) | Cache owns everything and is alive when called — no coupling (0) |
-| API surface | Op signatures stay clean | ~9 ops gain a defaulted `RetryContext` param; common call site unchanged (M, mechanical) |
+| API surface | Op signatures stay clean | Op signatures stay clean too — exclusion is a `cache.excluding(…)` bound view (`RetryView`), not a per-op param; retry path forwards to private `…Impl` (M, mechanical) |
 | Thread-safety of pool/topology | Required, and worse: retry fires from an arbitrary thread possibly long after the call (M–L) | Required, but it's ordinary concurrent method use (M) |
 
 **Rough estimate:** 1.b ≈ **40–55%** of 1.a's total effort. Savings concentrate in
@@ -311,19 +338,80 @@ lifetime-footgun and keeps the error trivially copyable/loggable/storable.
       (+ `ConnectionUsableAfterServerError` proves no stream desync).
 
 ### 11b — User-decided retry
-- [ ] `RetryContext` struct (§3.2), defaulted on every op signature.
-- [ ] Exclusion-aware `selectServerForKey` variant (skip `excludeNodes`).
-- [ ] Client-level `proxyToNonOwner` flag (default `true`, D5): gates the
-      owner→non-owner fallback; when `false`, throw `ownersExhausted=true`.
-- [ ] Populate `ownersExhausted` from `getOwners(key) ⊆ triedNodes`.
-- [ ] Thread-safety: guard `topology_` + `connectionPool_` + selection path
-      (retry may be driven concurrently with the read-loop's topology updates).
-- [ ] `triedNodes` accumulation across successive retries (union); the internal
-      BeforeSend sweep reports *all* nodes it touched.
-- [ ] Unit tests: exclusion selection; context threading; union accumulation;
-      proxyToNonOwner on/off; ownersExhausted.
-- [ ] Integration test: kill a node mid-run, user-loop retry lands on another.
-- [ ] Docs/example: the retry loop pattern (§3.2) in `examples/quickstart/`.
+- [x] `RetryContext` struct (§3.2) — **slice 1** (`include/hotrod/RetryContext.h`).
+      Op-level threading (slice 3) is done via a **bound view** (`RetryView` +
+      `cache.excluding()`), NOT a per-op parameter — base op signatures stay
+      retry-free. See the §3.2 design change and the slice-3 DECISIONS entry.
+- [x] Exclusion-aware `selectServerForKey` variant (skip `excludeNodes`) —
+      **slice 1**. Routing decision extracted to the pure, unit-tested
+      `orderKeyCandidates()` (`include/hotrod/ServerSelection.h`).
+- [x] Client-level `proxyToNonOwner` flag (default `true`, D5): gates the
+      owner→non-owner fallback; when `false`, throw `ownersExhausted=true` —
+      **slice 2** (`setProxyToNonOwner`/`getProxyToNonOwner`; gated in the pure
+      `orderKeyCandidates()`).
+- [x] Populate `ownersExhausted` from `getOwners(key) ⊆ triedNodes` —
+      **slice 1** (owner counts as exhausted if tried this attempt *or* excluded
+      from a prior one).
+- [x] Thread-safety: guard `topology_` + `connectionPool_` + selection path
+      (retry may be driven concurrently with the read-loop's topology updates) —
+      **slice 4**. One `stateMutex_` guards topology/hash/pool across
+      `handleTopologyUpdate` (read-loop) and `selectServerForKey` (user/retry);
+      pool + default connection are `shared_ptr` so a connection handed to an
+      in-flight op survives a concurrent topology eviction (no free
+      mid-`execute()`). Deadlock/lock-order verified; see the 2026-09-25 slice-4
+      DECISIONS entry.
+- [x] `triedNodes` accumulation across successive retries (union); the internal
+      BeforeSend sweep reports *all* nodes it touched. **Sweep reporting in slice
+      1** (`selectServerForKey` `triedOut` out-param); the cross-retry union
+      lands in **slice 3** — every op (first call included) folds
+      `excludeNodes ∪ tried ∪ e.triedNodes` into the thrown exception via the
+      pure `unionNodes()` helper (`ServerSelection.h`).
+- [x] Op-level retry threading via `RetryView` bound view — **slice 3**
+      (`include/hotrod/RetryView.h`; 9 private `…Impl` methods on `RemoteCache`;
+      `excluding(nodes)` / `excluding(exception)`).
+- [~] Unit tests: exclusion selection; union accumulation; proxyToNonOwner
+      on/off; ownersExhausted. **Exclusion selection (slice 1), proxyToNonOwner
+      on/off (slice 2), and union accumulation (slice 3) covered**
+      (`ServerSelectionTest`, 14 tests). `RetryView` dispatch is exercised by the
+      integration test below (needs live servers). **Keyless-op ordering for
+      `selectAnyServer` added in Step 11c slice 1** (`ServerSelectionTest`, now 17
+      tests).
+- [x] Integration test: kill a node mid-run, user-loop retry lands on another —
+      **slice 5** (`tests/integration/RetryViewIntegrationTest.cpp`, 3 tests):
+      `excluding()` route-around (deterministic), proxy-disabled owners-exhausted
+      throw (deterministic), and the catch→`excluding(e)` loop after killing the
+      primary owner. Verified 3/3 against a live 3-node cluster. See DECISIONS.md
+      (2026-09-25 slice-5 entry) for why the kill-loop test tolerates both the
+      throw-then-retry and the automatic re-routing recovery paths.
+- [x] Docs/example: the retry loop pattern (§3.2) in `examples/quickstart/` —
+      **slice 6** (`examples/quickstart/retry.cpp` + README section). Shows the
+      `getWithRetry`/`putWithRetry` loops (isTransient + outcomeUncertain +
+      `excluding(e)`) and states that `proxyToNonOwner` does not let a caller skip
+      the catch block. Builds warning-clean; runs end-to-end against a live server.
+
+### Step 11c — keyless-op retry (ping first)
+
+Keyless operations (no routing key/owner) get the same two-tier model: automatic
+before-send failover across all servers, user-decided after-send retry.
+
+- [x] `selectAnyServer(ctx, triedOut)` + `pingImpl(ctx)` + `RetryView::ping()` —
+      **slice 1** (`src/operations/RemoteCache.cpp`, `RemoteCache.h`,
+      `RetryView.h`). Routes via `orderKeyCandidates({}, allServers, exclude)`
+      (empty owners → all servers in topology order), sweeps to the next reachable
+      server before send, falls back to the seed connection when no topology is
+      known yet, and throws `BeforeSend`/`ownersExhausted=false` on exhaustion.
+      Unit: 3 keyless-ordering tests in `ServerSelectionTest`.
+- [x] Integration test: keyless `excluding().ping()` route-around; exclude-all
+      throws BeforeSend with `ownersExhausted=false`; catch→`excluding(e).ping()`
+      recovers after the first-candidate node is killed — **slice 2**
+      (`tests/integration/PingRetryIntegrationTest.cpp`, 3 tests, verified 3/3).
+- [x] Docs/example: `ping` retry in `examples/quickstart/retry.cpp` (a keyless
+      `pingWithRetry` loop) + README/DECISIONS — **slice 3**.
+
+**Divergence note:** Java's user-facing `CachePingOperation` (`supportRetry=true`)
+routes via the keyless round-robin balancer and auto-retries up to `maxRetries`;
+we use deterministic topology order and keep after-send retry user-decided (D2).
+`ownersExhausted` is always `false` for keyless ops (no owner concept).
 
 ---
 
diff --git a/docs/STATUS.md b/docs/STATUS.md
index 18ce6a5..f32b265 100644
--- a/docs/STATUS.md
+++ b/docs/STATUS.md
@@ -33,20 +33,59 @@ _(Full per-session workflow: [`WORKFLOW.md`](WORKFLOW.md).)_
 
 _The 1–3 concrete things to do next. Keep this short and current._
 
-1. **Step 11a is complete.** Typed `HotRodClientException` (pure-data:
-   `FailurePhase` + `serverStatus` + `triedNodes` + `ownersExhausted`) now
-   replaces bare `std::runtime_error` across the ops, and the server ERROR
-   response (opcode 0x50) is parsed and surfaced with its status + message (see
-   "Working and shipped"). Next roadmap step: **Step 11b — user-decided retry**
-   (defaulted `RetryContext` on the ops, exclusion-aware `selectServerForKey`,
-   `proxyToNonOwner` default true, thread-safe pool/topology). Retry policy /
-   idempotency stays the user's call. Full design + progress checklist:
-   [`ERROR_HANDLING_DESIGN.md`](ERROR_HANDLING_DESIGN.md); rationale in
-   [`DECISIONS.md`](DECISIONS.md) (2026-09-21 and 2026-09-25 entries). **Start
-   with the 11b checklist.**
-2. **Benchmark the multiplexing path** — the async rewrite targets 5–10×
+1. **Step 11c COMPLETE — keyless-op retry (ping) shipped (slices 1–3).** Keyless
+   operations now mirror the keyed retry strategy: automatic before-send failover
+   across all servers, user-decided after-send retry (D2). `ping()` routes through
+   `selectAnyServer(ctx, triedOut)`, which orders candidates via
+   `orderKeyCandidates({}, allServers, exclude)` (empty owners → all servers in
+   topology order), sweeps to the next server when one is unreachable, and — when
+   no topology is known yet (ping is often the very first op) — falls back to the
+   seed connection. `RetryView::ping()` forwards the exclusion set, so
+   `cache.excluding(e).ping()` works exactly like keyed ops; `ownersExhausted` is
+   always `false` for keyless ops. (1) `pingImpl` + `selectAnyServer` in
+   `RemoteCache.cpp`, 3 keyless-ordering unit tests; (2) `PingRetryIntegrationTest`
+   (3 tests) — `excluding()` routes the keyless ping to another server, excluding
+   all servers throws BeforeSend with `ownersExhausted=false`, and the
+   catch→`excluding(e).ping()` loop recovers after the first-candidate node is
+   killed; (3) `examples/quickstart/retry.cpp` gained a keyless `pingWithRetry`
+   loop + README/ERROR_HANDLING docs. The implementation is generic, so future
+   non-key ops (server stats/admin) reuse `selectAnyServer`. Rationale in
+   [`DECISIONS.md`](DECISIONS.md) (2026-09-25 Step 11c entries). **Next: pick a new
+   roadmap step — Step 12 (bulk ops) or benchmark the multiplexing path.**
+2. **Step 11b COMPLETE — user-decided retry shipped (slices 1–6).** (1) `selectServerForKey` routes
+   over the pure, unit-tested `orderKeyCandidates()` helper (owners → non-owner
+   fallback, minus an exclusion set), takes a defaulted `RetryContext`, reports
+   the nodes it tried via an out-param, and computes `ownersExhausted` honestly.
+   (2) Client-level `proxyToNonOwner` flag (default `true`, D5:
+   `setProxyToNonOwner`/`getProxyToNonOwner`) gates the owner→non-owner fallback;
+   when `false`, exhausting owners throws a transient exception with
+   `ownersExhausted=true` instead of proxying. (3) Op-level retry threading via a
+   **bound-view** API — base op signatures stay retry-free; `cache.excluding(nodes)`
+   / `cache.excluding(exception)` returns a `RetryView` that forwards to private
+   `…Impl(args, const RetryContext&)` methods, and every op (including the plain
+   first call) unions the nodes it tried into the thrown exception's `triedNodes`,
+   so a caught exception feeds straight back as the next attempt's exclusion set.
+   New public headers: `RetryContext.h`, `ServerSelection.h`, `RetryView.h`.
+   (4) Thread-safety: a single `stateMutex_` guards `topology_` +
+   `consistentHash_` + `connectionPool_` across the read-loop's topology callback
+   (`handleTopologyUpdate`) and user/retry threads (`selectServerForKey` →
+   `getConnectionForServer`); the connection pool holds `shared_ptr` and each op
+   keeps its connection alive for the operation's duration, so a concurrent
+   topology update can't free a connection mid-`execute()`.
+   (5) End-to-end integration coverage: `RetryViewIntegrationTest` (3 tests)
+   exercises `excluding()` dispatch — routes around an excluded owner, throws
+   `ownersExhausted` when proxy is disabled and all owners are excluded, and runs
+   the documented catch→`excluding(e)` retry loop after killing the primary owner.
+   (6) Retry-loop example `examples/quickstart/retry.cpp` (+ README section) —
+   `getWithRetry`/`putWithRetry` loops using `isTransient`/`outcomeUncertain`/
+   `excluding(e)`, and the clarification that `proxyToNonOwner` does not let a
+   caller skip the catch block. Retry policy/idempotency stays the user's call.
+   Full design + progress checklist:
+   [`ERROR_HANDLING_DESIGN.md`](ERROR_HANDLING_DESIGN.md) §5; rationale in
+   [`DECISIONS.md`](DECISIONS.md) (2026-09-21, 2026-09-25 entries).
+3. **Benchmark the multiplexing path** — the async rewrite targets 5–10×
    concurrent throughput; this has not been measured yet.
-3. **Small cleanup:** read header "other params" when `paramCount > 0`
+4. **Small cleanup:** read header "other params" when `paramCount > 0`
    (two `TODO`s in `src/operations/RemoteCache.cpp:140,182`).
 
 ---
@@ -67,8 +106,8 @@ pull from here next. Step numbers follow
 
   Java ref: `GetWithMetadataOperation`, `ReplaceIfUnmodifiedOperation`. See the
   [step-by-step plan](#plan-metadata-operations-step-10) below.
-- [ ] **Step 11 — Error handling.** Design agreed 2026-09-21 →
-  [`ERROR_HANDLING_DESIGN.md`](ERROR_HANDLING_DESIGN.md). Split:
+- [x] **Step 11 — Error handling — COMPLETE (11a + 11b + 11c).** Design agreed
+  2026-09-21 → [`ERROR_HANDLING_DESIGN.md`](ERROR_HANDLING_DESIGN.md). Split:
   - [x] **11a — Error surfacing** — **shipped 2026-09-25** (11 unit + 3
     integration tests). ERROR response parsing (opcode 0x50) with
     length-prefixed message drained in the read loop (keeps the stream in sync
@@ -79,10 +118,32 @@ pull from here next. Step numbers follow
     classification helpers: `isTransient()` (futility — library's call) and
     `outcomeUncertain()` (ambiguity — 0x86/AfterSend). `COMMAND_TIMEOUT` (0x86)
     is transient **and** outcome-uncertain (intentionally diverges from Java).
-  - [ ] **11b — User-decided retry:** defaulted `RetryContext` on the ops
-    (fork 1.b), exclusion-aware `selectServerForKey`, `proxyToNonOwner` default
-    true, thread-safe pool/topology. Retry policy/idempotency is the user's
-    call, not automatic.
+  - [x] **11b — User-decided retry** — **shipped 2026-09-25**: exclusion-aware
+    `selectServerForKey`, `proxyToNonOwner` default true, a bound-view retry API
+    (fork 1.b), thread-safe pool/topology. Retry policy/idempotency is the
+    user's call, not automatic. **Slices 1–3:** exclusion-aware selection
+    via the pure `orderKeyCandidates()` helper (`ServerSelection.h`) +
+    `RetryContext` (`RetryContext.h`); client-level `proxyToNonOwner` flag
+    (default `true`); op-level retry threading via `RetryView`/`cache.excluding()`
+    (`RetryView.h`) with `triedNodes` unioned into every thrown exception (pure
+    `unionNodes()` helper). Base op signatures stay retry-free. **Slice 4
+    (thread-safety) done:** `stateMutex_` guards topology/hash/pool across the
+    read-loop and user/retry threads; pool is `shared_ptr` and each op keeps its
+    connection alive for its duration (no free mid-`execute()`). **Slice 5
+    (integration test):** `RetryViewIntegrationTest` (3 tests) covers `excluding()`
+    route-around, proxy-disabled owners-exhausted throw, and the catch→retry loop
+    after killing the primary owner. **Slice 6 (example):**
+    `examples/quickstart/retry.cpp` shows the retry loop and the `proxyToNonOwner`
+    clarification.
+  - [x] **11c — Keyless-op retry (ping)** — **shipped 2026-09-25** (3 unit + 3
+    integration tests). Keyless ops mirror the keyed two-tier model: automatic
+    before-send failover across all servers via `selectAnyServer` (deterministic
+    topology order, minus an exclusion set; seed-connection fallback when no
+    topology is known yet), user-decided after-send retry via
+    `cache.excluding(e).ping()`. `ownersExhausted` is always `false` for keyless
+    ops; `selectAnyServer` is generic for future non-key ops.
+    `PingRetryIntegrationTest` (3 tests) + a keyless `pingWithRetry` in the retry
+    example.
   Java ref: `org.infinispan.client.hotrod.exceptions.*`.
 - [ ] **Step 12 — Bulk operations.** `GET_ALL` (0x2F), `PUT_ALL` (0x2D),
   `BULK_GET` (0x1F, iterator-style).
@@ -116,6 +177,11 @@ multiplexing. See "Working and shipped" below._
 - Hash-aware routing → primary owner, with automatic failover
 - Connection pooling (one connection per server)
 - **Full CRUD:** PING / GET / PUT / REMOVE
+- **Keyless-op retry (Step 11c):** `ping()` routes through `selectAnyServer` —
+  automatic before-send failover across all servers (topology order, minus an
+  exclusion set), seed-connection fallback when no topology is known yet, and
+  user-decided after-send retry via `cache.excluding(e).ping()`. Same strategy
+  intended for all future non-key ops; `ownersExhausted` always `false`
 - **getWithMetadata** (0x1B/0x1C) — value + entry version/expiration metadata,
   the entry point for version-based CAS (Step 10)
 - **removeWithVersion** (0x0D/0x0E) — version-based conditional remove (CAS);
@@ -146,14 +212,28 @@ multiplexing. See "Working and shipped" below._
   portability and `-Werror` build parity also landed (Sept 2026).
 
 **Test status (verified 2026-09-25):**
-- Unit: **187/187** passing (`./build/unit_tests`, <1s)
-- Integration: **78/78** passing across 15 suites (`ctest`, spins up Docker
+- Unit: **204/204** passing (`./build/unit_tests`, <1s) — +17 for
+  `ServerSelectionTest` (`orderKeyCandidates` + `unionNodes`, Step 11b slices 1–3;
+  +3 keyless-ordering tests for `selectAnyServer`, Step 11c slice 1)
+- Integration: **84/84** passing across 17 suites (`ctest`, spins up Docker
   Infinispan single-server + multi-node clusters), now also green on Linux CI.
+  +3 for `RetryViewIntegrationTest` (Step 11b: `excluding()` routes around an
+  owner; proxy-disabled owners-exhausted throw; the catch→`excluding(e)` retry
+  loop recovers after the primary owner is killed) — verified 3/3 this session.
+  +3 for `PingRetryIntegrationTest` (Step 11c: keyless `excluding().ping()` routes
+  to another server; excluding all servers throws BeforeSend with
+  `ownersExhausted=false`; the catch→`excluding(e).ping()` loop recovers after the
+  first-candidate node is killed) — verified 3/3 this session.
   The interlaced distributed tests (`ConcurrentMultiServerTest`) were fixed to
   tolerate a GET racing ahead of its PUT — a `nullopt` is expected, only a
   present-but-wrong value is an error.
 - Full run: `ctest --test-dir build --output-on-failure` → 100% pass
-  (16 ctest tests: 1 unit + 15 integration suites)
+  (18 ctest tests: 1 unit + 17 integration suites)
+- **Concurrency validated (2026-09-25):** the concurrent suites pass 8/8
+  (incl. `ConcurrentWithFailover`), and a ThreadSanitizer run (`build-tsan/`)
+  reports **no races in production code** — slice-4's `stateMutex_`/`shared_ptr`
+  synchronization held clean. A test-harness race in `TopologyTestFixture.h`
+  `createClient()` (concurrent `push_back`) was fixed with a `clientsMutex`.
 - **Local caveat:** the multi-node cluster suites bind fixed host ports
   `11222/11322/11422/11522`; free `11222` (e.g. stop the `memory-service`
   Infinispan) before running them locally, or they fail with "port is already
@@ -169,6 +249,7 @@ full list (Steps 10–12, benchmarks, TLS, code TODOs).
 | What | Where |
 |------|-------|
 | Public API | `include/hotrod/RemoteCache.h` |
+| Retry API (bound view) | `include/hotrod/RetryView.h`, `RetryContext.h`; routing helpers in `ServerSelection.h` |
 | Async transport core | `src/transport/MultiplexedConnection.cpp` |
 | Operations (PING/GET/PUT/REMOVE) | `src/operations/RemoteCache.cpp` |
 | Codecs | `src/codec/`, `src/hash/`, `src/auth/`, `src/topology/` |
diff --git a/docs/WORKFLOW.md b/docs/WORKFLOW.md
index f813d38..7acc038 100644
--- a/docs/WORKFLOW.md
+++ b/docs/WORKFLOW.md
@@ -68,8 +68,11 @@ normal case. Split what you write by *kind*, so each doc keeps one job:
    gotchas discovered, schema/protocol bugs, and design decisions. If future-you
    would want to know why or how something happened, it goes here — not in
    STATUS.md. **Never rewrite old entries**; supersede with a new dated one.
-3. **`README.md`** — update **only if a user-facing fact changed** (a feature
-   shipped, test numbers, public API). Most sessions won't touch it.
+3. **`README.md`** — **check it after every commit.** The README carries a subset
+   of user-facing facts (test counts, shipped features, step/milestone status,
+   public API). If the commit changed any of them, update the README in the *same*
+   commit so it never drifts from STATUS.md. If nothing user-facing changed, a
+   quick glance confirms it and you move on — but the check itself is not optional.
 4. **Commit.** The `pre-commit` hook will block you if source changed and
    STATUS.md wasn't staged (see the rule box at the top of this file).
 
diff --git a/examples/quickstart/CMakeLists.txt b/examples/quickstart/CMakeLists.txt
index c00f33e..03eba97 100644
--- a/examples/quickstart/CMakeLists.txt
+++ b/examples/quickstart/CMakeLists.txt
@@ -22,17 +22,25 @@ endif()
 # Include directories
 include_directories(${HOTROD_INCLUDE_DIR})
 
-# Create executable
+# Create executables
 add_executable(quickstart quickstart.cpp)
 
+# User-decided retry example (Step 11b)
+add_executable(retry retry.cpp)
+
 # Link against Hot Rod client library
 target_link_libraries(quickstart
     ${HOTROD_LIB_DIR}/libhotrod-client.a
     pthread
 )
 
-# Install target (optional)
-install(TARGETS quickstart DESTINATION bin)
+target_link_libraries(retry
+    ${HOTROD_LIB_DIR}/libhotrod-client.a
+    pthread
+)
+
+# Install targets (optional)
+install(TARGETS quickstart retry DESTINATION bin)
 
 message(STATUS "Hot Rod Quickstart Example")
 message(STATUS "  Include dir: ${HOTROD_INCLUDE_DIR}")
diff --git a/examples/quickstart/README.md b/examples/quickstart/README.md
index 9bb7f0f..1c4c4c0 100644
--- a/examples/quickstart/README.md
+++ b/examples/quickstart/README.md
@@ -125,10 +125,84 @@ cache.remove(key);
 cache.disconnect();
 ```
 
+## User-Decided Retry (`retry` example)
+
+`retry.cpp` demonstrates Step 11b/11c — **the client never retries on its own**.
+A failure surfaces as a typed `HotRodClientException` carrying the facts you need
+(which phase it failed in, which nodes were already tried); you opt into a retry
+explicitly, on the failure path, via `cache.excluding(e)`:
+
+```cpp
+std::vector<ServerAddress> excluded;      // nodes tried so far
+for (int attempt = 1; ; ++attempt) {
+    try {
+        return excluded.empty()
+            ? cache.get(key).get()
+            : cache.excluding(excluded).get(key).get();
+    } catch (const HotRodClientException& e) {
+        if (!isTransient(e)) throw;       // permanent — give up
+        // ... your idempotency decision (outcomeUncertain(e)) for writes ...
+        excluded = e.triedNodes;          // avoid tried nodes next time
+    }
+}
+```
+
+Why user-decided? Only the caller knows whether replaying *this* operation is
+safe. `isTransient(e)` answers "is a retry worthwhile?"; `outcomeUncertain(e)`
+answers "might it already have applied?" — you `AND` the second with your own
+knowledge of the op before replaying a non-idempotent write.
+
+> **The `proxyToNonOwner` flag does NOT let you skip the catch block.**
+> `proxyToNonOwner=true` (the default) only widens the *connection-selection*
+> candidate pool for a **single** dispatch: if no owner is reachable *before the
+> request is sent*, selection falls through to another node that proxies to the
+> owner. But one operation still executes on exactly **one** node, so any failure
+> *after* the request is sent (dropped connection, server ERROR, command timeout)
+> still comes back to you as an exception. The flag changes where the first
+> attempt is *sent*; it does not make retry automatic.
+
+**Keyless operations retry the same way (Step 11c).** `ping()` has no key and so
+no owners; it routes to any server in topology order, automatically fails over to
+the next server *before send*, and surfaces after-send/server errors to you just
+like a keyed op — so you retry it with `cache.excluding(e).ping()`:
+
+```cpp
+std::vector<ServerAddress> excluded;
+for (int attempt = 1; ; ++attempt) {
+    try {
+        if (excluded.empty()) cache.ping().get();
+        else                  cache.excluding(excluded).ping().get();
+        return;
+    } catch (const HotRodClientException& e) {
+        if (!isTransient(e)) throw;   // ping is idempotent — always safe to replay
+        excluded = e.triedNodes;
+    }
+}
+```
+
+The only visible difference from a keyed op: a keyless exhaustion reports
+`ownersExhausted == false` (there are no owners). The example calls
+`pingWithRetry()` right after connecting, since `ping` is typically the first op —
+often before any topology has arrived, in which case it falls back to the seed
+connection.
+
+Build it alongside `quickstart` (both targets are in `CMakeLists.txt`):
+
+```bash
+cd build && cmake .. && make        # builds quickstart and retry
+./retry
+```
+
+Run it against a **cluster** and kill an owner node mid-run to actually exercise
+the retry path (the same scenarios are covered by
+`tests/integration/RetryViewIntegrationTest.cpp` for keyed ops and
+`tests/integration/PingRetryIntegrationTest.cpp` for keyless `ping`).
+
 ## Files
 
-- `quickstart.cpp` - Main example code
-- `CMakeLists.txt` - Build configuration
+- `quickstart.cpp` - Basic PUT/GET/REMOVE example
+- `retry.cpp` - User-decided retry loop, keyed + keyless `ping` (Step 11b/11c)
+- `CMakeLists.txt` - Build configuration (builds both examples)
 - `start-server.sh` - Script to start Infinispan server
 - `stop-server.sh` - Script to stop Infinispan server
 - `README.md` - This file
diff --git a/examples/quickstart/retry.cpp b/examples/quickstart/retry.cpp
new file mode 100644
index 0000000..de55da8
--- /dev/null
+++ b/examples/quickstart/retry.cpp
@@ -0,0 +1,214 @@
+/**
+ * Hot Rod C++ Client - User-Decided Retry Example (Step 11b / 11c)
+ *
+ * The base operations (cache.get/put/...) never retry on their own: a failure
+ * surfaces as a typed HotRodClientException carrying the facts you need to decide
+ * what to do next (which phase it failed in, which nodes were already tried).
+ * You opt into a retry explicitly, on the failure path, via cache.excluding():
+ *
+ *     try {
+ *         value = cache.get(key).get();
+ *     } catch (const HotRodClientException& e) {
+ *         if (!isTransient(e)) throw;                 // futile — give up
+ *         // ... your idempotency decision (see below) ...
+ *         value = cache.excluding(e).get(key).get();  // retry, avoiding e.triedNodes
+ *     }
+ *
+ * Why the library does not retry for you: only YOU know whether replaying THIS
+ * particular operation is safe. See the two-axis model below and D2/D4 in
+ * docs/DECISIONS.md / docs/ERROR_HANDLING_DESIGN.md.
+ *
+ * IMPORTANT — the proxyToNonOwner flag does NOT let you skip the catch block.
+ * ------------------------------------------------------------------------------
+ * proxyToNonOwner=true (the default) only widens the *connection-selection*
+ * candidate pool for a SINGLE dispatch: if none of a key's owners can be reached
+ * *before the request is sent*, selection falls through to another node that
+ * proxies to the real owner. But one operation still executes on exactly ONE
+ * node. So a failure that happens *after* the request was sent (a dropped
+ * connection, a server ERROR, a command timeout) still comes back to you as an
+ * exception. The flag changes where the first attempt is *sent*; it does not make
+ * retry automatic and does not remove the need to catch and retry yourself.
+ *
+ * KEYLESS operations follow the SAME pattern (Step 11c).
+ * ------------------------------------------------------------------------------
+ * ping() has no key, so it has no owners — it routes to any server (in topology
+ * order), automatically failing over to the next server before send, and
+ * surfacing after-send/server errors to you exactly like a keyed op. You retry it
+ * the same way, via cache.excluding(e).ping(). The only visible difference is that
+ * a keyless exhaustion reports ownersExhausted == false (there are no owners). See
+ * pingWithRetry() below.
+ */
+
+#include "hotrod/RemoteCache.h"
+#include "hotrod/HotRodClientException.h"
+#include <iostream>
+#include <optional>
+#include <string>
+#include <vector>
+
+using namespace hotrod;
+
+// A reusable, idempotent-GET retry loop that follows the documented pattern.
+//
+// Each failed attempt reports the nodes it tried in e.triedNodes; feeding that
+// back through cache.excluding(e) makes the next attempt avoid them, so the
+// exclusion set grows monotonically until a live node serves the key or we run
+// out of places to try. GET is idempotent, so replaying is always safe here — a
+// non-idempotent op (see putWithRetry below) needs the extra outcomeUncertain()
+// check before it may be replayed.
+std::optional<ByteArray> getWithRetry(RemoteCache& cache, const ByteArray& key,
+                                      int maxAttempts = 4) {
+    std::vector<ServerAddress> excluded;  // nodes tried so far (empty on 1st try)
+
+    for (int attempt = 1; ; ++attempt) {
+        try {
+            // First attempt is the plain, retry-free call; later attempts opt
+            // into exclusion seeded from the previous failure.
+            return excluded.empty()
+                ? cache.get(key).get()
+                : cache.excluding(excluded).get(key).get();
+        } catch (const HotRodClientException& e) {
+            std::cerr << "  attempt " << attempt << " failed: " << e.what()
+                      << "  (tried " << e.triedNodes.size() << " node(s))\n";
+
+            // Axis 2 — is a retry even worthwhile? A permanent, request-level
+            // error (bad request, cache not found, ...) will fail identically on
+            // every node, so stop.
+            if (!isTransient(e)) {
+                std::cerr << "  -> permanent failure, not retrying\n";
+                throw;
+            }
+            // No live node left to try (all owners exhausted and no proxy).
+            if (attempt >= maxAttempts) {
+                std::cerr << "  -> giving up after " << attempt << " attempts\n";
+                throw;
+            }
+            // Carry the tried nodes forward as the next attempt's exclusion set.
+            excluded = e.triedNodes;
+        }
+    }
+}
+
+// A keyless retry loop (Step 11c). ping() has no key/owner, so selection tries
+// every server in topology order; the loop is identical to getWithRetry() — plain
+// first call, then cache.excluding(e).ping() seeded from the failure. ping is
+// naturally idempotent, so it is always safe to replay. On exhaustion the thrown
+// exception reports ownersExhausted == false (a keyless op has no owners).
+void pingWithRetry(RemoteCache& cache, int maxAttempts = 4) {
+    std::vector<ServerAddress> excluded;
+
+    for (int attempt = 1; ; ++attempt) {
+        try {
+            if (excluded.empty())
+                cache.ping().get();
+            else
+                cache.excluding(excluded).ping().get();
+            return;
+        } catch (const HotRodClientException& e) {
+            std::cerr << "  ping attempt " << attempt << " failed: " << e.what()
+                      << "  (tried " << e.triedNodes.size() << " node(s))\n";
+
+            if (!isTransient(e)) throw;
+            if (attempt >= maxAttempts) throw;
+            excluded = e.triedNodes;
+        }
+    }
+}
+
+// The same loop for a NON-idempotent write. The extra rule: if the outcome is
+// uncertain (the request may have been applied before the failure — an AfterSend
+// drop or a command timeout), replaying a put could double-apply. Here we still
+// retry because put(key, value) is naturally idempotent (same key+value), but the
+// point is that THIS is the decision only the caller can make.
+std::optional<EntryWithMetadata> putWithRetry(RemoteCache& cache,
+                                              const ByteArray& key,
+                                              const ByteArray& value,
+                                              int maxAttempts = 4) {
+    std::vector<ServerAddress> excluded;
+
+    for (int attempt = 1; ; ++attempt) {
+        try {
+            return excluded.empty()
+                ? cache.put(key, value).get()
+                : cache.excluding(excluded).put(key, value).get();
+        } catch (const HotRodClientException& e) {
+            std::cerr << "  put attempt " << attempt << " failed: " << e.what() << "\n";
+
+            if (!isTransient(e)) throw;
+
+            if (outcomeUncertain(e)) {
+                // The write may already have landed on the server. For a plain
+                // idempotent overwrite that is fine; for an accumulating or
+                // conditional op you might instead verify with getWithMetadata()
+                // before replaying. Decide per operation — the library will not.
+                std::cerr << "  -> outcome uncertain; safe to replay this "
+                             "idempotent put\n";
+            }
+            if (attempt >= maxAttempts) throw;
+            excluded = e.triedNodes;
+        }
+    }
+}
+
+int main() {
+    std::cout << "=== Hot Rod C++ Client - User-Decided Retry ===\n";
+
+    try {
+        RemoteCache cache("localhost", 11222, "quickstart-cache");
+
+        // Hash-distribution awareness is what gives the client owner-aware
+        // routing (and therefore a meaningful set of nodes to exclude on retry).
+        cache.setClientIntelligence(ClientIntelligence::HASH_DISTRIBUTION_AWARE);
+
+        // proxyToNonOwner defaults to true; shown here for emphasis. Remember: it
+        // only affects where the FIRST attempt is sent — it does not make retry
+        // automatic (see the file header).
+        cache.setProxyToNonOwner(true);
+
+        std::cout << "Connecting...\n";
+        cache.connect();
+        std::cout << "Connected.\n\n";
+
+        // Keyless op: ping is usually one of the first calls, before any topology
+        // has arrived; it routes to any server and fails over the same way.
+        std::cout << "--- PING with retry (keyless) ---\n";
+        pingWithRetry(cache);
+        std::cout << "PING ok\n\n";
+
+        std::string keyStr = "retry-demo-key";
+        std::string valueStr = "retry-demo-value";
+        ByteArray key(keyStr.begin(), keyStr.end());
+        ByteArray value(valueStr.begin(), valueStr.end());
+
+        std::cout << "--- PUT with retry ---\n";
+        putWithRetry(cache, key, value);
+        std::cout << "PUT ok: " << keyStr << " = " << valueStr << "\n\n";
+
+        std::cout << "--- GET with retry ---\n";
+        auto result = getWithRetry(cache, key);
+        if (result.has_value()) {
+            std::string got(result->begin(), result->end());
+            std::cout << "GET ok: " << keyStr << " = " << got << "\n";
+        } else {
+            std::cout << "GET: <not found>\n";
+        }
+
+        std::cout << "\nTo actually exercise the retry path, run against a cluster\n"
+                     "and kill an owner node mid-run: the plain call throws a\n"
+                     "HotRodClientException and the loop retries via excluding(e).\n";
+
+        cache.disconnect();
+        std::cout << "\nDisconnected.\n";
+
+    } catch (const HotRodClientException& e) {
+        // Typed catch: still catchable as std::runtime_error if you prefer.
+        std::cerr << "Hot Rod error: " << e.what() << "\n";
+        return 1;
+    } catch (const std::exception& e) {
+        std::cerr << "Error: " << e.what() << "\n";
+        return 1;
+    }
+
+    std::cout << "\n=== Retry Example Complete ===\n";
+    return 0;
+}
diff --git a/include/hotrod/RemoteCache.h b/include/hotrod/RemoteCache.h
index 1b7b7bb..e13b5fd 100644
--- a/include/hotrod/RemoteCache.h
+++ b/include/hotrod/RemoteCache.h
@@ -5,15 +5,19 @@
 #include "MultiplexedConnection.h"
 #include "HeaderCodec.h"
 #include "ConsistentHash.h"
+#include "RetryContext.h"
 #include <string>
 #include <memory>
 #include <cstdint>
 #include <map>
 #include <future>
 #include <optional>
+#include <mutex>
 
 namespace hotrod {
 
+class RetryView;  // bound exclusion-aware view; defined in RetryView.h
+
 /**
  * High-level Hot Rod client API.
  *
@@ -319,42 +323,142 @@ class RemoteCache {
         return clientIntelligence_;
     }
 
+    /**
+     * Control non-owner proxy fallback for hash-aware routing (Step 11b, D5).
+     *
+     * When true (the default, matching the Java client), if none of a key's
+     * owners can be reached, selection falls through to any other server in the
+     * topology, which proxies the request to the real owner. When false,
+     * selection stops once the owners are exhausted and throws a transient
+     * HotRodClientException with ownersExhausted=true, letting the caller decide
+     * whether to drive the proxy step itself.
+     *
+     * This governs the BeforeSend connection-fallback in selectServerForKey
+     * only; it does not make retry automatic (retry stays user-decided, D2).
+     */
+    void setProxyToNonOwner(bool proxyToNonOwner) {
+        proxyToNonOwner_ = proxyToNonOwner;
+    }
+
+    /**
+     * Whether non-owner proxy fallback is enabled (default true).
+     */
+    bool getProxyToNonOwner() const {
+        return proxyToNonOwner_;
+    }
+
     /**
      * Get current topology information.
      * Only populated if client intelligence is TOPOLOGY_AWARE or HASH_DISTRIBUTION_AWARE.
+     *
+     * Thread-safety: returns a reference to internal state that the read-loop
+     * thread mutates under stateMutex_ on topology updates. It is a
+     * testing/debugging accessor and must NOT be read concurrently with live
+     * operations or topology churn. Use getTopologyId() (locked, by value) for a
+     * safe point-in-time read.
      */
     const TopologyInfo& getTopology() const {
         return topology_;
     }
 
     /**
-     * Get current topology ID.
+     * Get current topology ID (thread-safe: read under stateMutex_).
      */
     VInt getTopologyId() const {
+        std::lock_guard<std::mutex> lock(stateMutex_);
         return topology_.getTopologyId();
     }
 
     /**
      * Get the ConsistentHash instance (for testing/debugging).
+     *
+     * Thread-safety: same caveat as getTopology() — returns a reference to state
+     * the read-loop mutates under stateMutex_; not safe to read concurrently with
+     * live operations or topology churn.
      */
     const ConsistentHash& getConsistentHash() const {
         return consistentHash_;
     }
 
+    /**
+     * Return a bound view of this cache that excludes the given nodes from
+     * routing (Step 11b). The base operations stay retry-free; a caller opts into
+     * exclusion only on the retry path, seeding it from a caught exception:
+     *
+     *   try { return cache.get(key).get(); }
+     *   catch (const HotRodClientException& e) {
+     *       if (!isTransient(e)) throw;              // + your idempotency call
+     *       return cache.excluding(e).get(key).get();  // avoids e.triedNodes
+     *   }
+     *
+     * The returned RetryView is non-owning and must not outlive this cache.
+     */
+    RetryView excluding(std::vector<ServerAddress> excludeNodes);
+
+    /**
+     * Convenience overload: exclude exactly the nodes a prior attempt tried,
+     * as reported by the exception (uses e.triedNodes).
+     */
+    RetryView excluding(const HotRodClientException& e);
+
 private:
+    friend class RetryView;  // forwards to the *Impl methods below
+
+    // Exclusion-aware implementations behind the public retry-free operations.
+    // The public ops call these with an empty RetryContext; a RetryView calls
+    // them with its bound context. Each surfaces failures as a
+    // HotRodClientException whose triedNodes is the union of the context's
+    // excludeNodes and the nodes this attempt touched (see RemoteCache.cpp).
+    // Keyless operation (no routing object). Auto-fails over across all servers
+    // before send (selectAnyServer); after-send/server errors surface to the
+    // caller with triedNodes, retryable via a RetryView, exactly like keyed ops.
+    std::future<void> pingImpl(const RetryContext& ctx);
+    std::future<std::optional<ByteArray>> getImpl(const ByteArray& key, const RetryContext& ctx);
+    std::future<std::optional<EntryWithMetadata>> getWithMetadataImpl(const ByteArray& key, const RetryContext& ctx);
+    std::future<std::optional<EntryWithMetadata>> putImpl(const ByteArray& key, const ByteArray& value,
+                                                          uint64_t lifespan, uint64_t maxIdle, bool previousValue,
+                                                          const RetryContext& ctx);
+    std::future<std::optional<EntryWithMetadata>> putIfAbsentImpl(const ByteArray& key, const ByteArray& value,
+                                                                  uint64_t lifespan, uint64_t maxIdle, bool previousValue,
+                                                                  const RetryContext& ctx);
+    std::future<std::optional<EntryWithMetadata>> replaceImpl(const ByteArray& key, const ByteArray& value,
+                                                              uint64_t lifespan, uint64_t maxIdle, bool previousValue,
+                                                              const RetryContext& ctx);
+    std::future<bool> containsKeyImpl(const ByteArray& key, const RetryContext& ctx);
+    std::future<std::optional<EntryWithMetadata>> removeImpl(const ByteArray& key, bool previousValue,
+                                                             const RetryContext& ctx);
+    std::future<bool> removeWithVersionImpl(const ByteArray& key, int64_t version, const RetryContext& ctx);
+    std::future<bool> replaceWithVersionImpl(const ByteArray& key, const ByteArray& value,
+                                             int64_t version, uint64_t lifespan, uint64_t maxIdle,
+                                             const RetryContext& ctx);
+
     std::string host_;
     uint16_t port_;
     std::string cacheName_;
-    std::unique_ptr<MultiplexedConnection> connection_;
+    // shared_ptr (not unique_ptr) so a connection handed to an in-flight
+    // operation stays alive even if a concurrent topology update evicts it from
+    // the pool (see selectServerForKey / cleanupStaleConnections, Step 11b slice 4).
+    std::shared_ptr<MultiplexedConnection> connection_;
     // Note: messageIdCounter_ removed - MultiplexedConnection generates IDs
     uint8_t protocolVersion_;  // Protocol version (default: VERSION_41)
     ClientIntelligence clientIntelligence_;  // Client intelligence level
+    bool proxyToNonOwner_ = true;  // Owner->non-owner fallback in routing (D5)
+
+    // stateMutex_ guards the routing/pool state below (topology_, consistentHash_,
+    // connectionPool_) against concurrent access by the read-loop thread (topology
+    // callbacks → handleTopologyUpdate) and user/retry threads (ops →
+    // selectServerForKey → getConnectionForServer). Mutable so the const
+    // getTopologyId() accessor can lock it. Lock discipline: the two public entry
+    // points (handleTopologyUpdate, selectServerForKey) acquire it at the top;
+    // their internal helpers (cleanupStaleConnections, getConnectionForServer)
+    // assume it is already held and must not re-lock.
+    mutable std::mutex stateMutex_;
     TopologyInfo topology_;  // Current cluster topology
     ConsistentHash consistentHash_;  // Hash-aware routing (for HASH_DISTRIBUTION_AWARE)
 
     // Connection pool: one MultiplexedConnection per server
     // Key = "host:port", Value = MultiplexedConnection instance
-    std::map<std::string, std::unique_ptr<MultiplexedConnection>> connectionPool_;
+    std::map<std::string, std::shared_ptr<MultiplexedConnection>> connectionPool_;
 
     /**
      * Handle topology update callback from MultiplexedConnection.
@@ -371,24 +475,71 @@ class RemoteCache {
      *
      * Otherwise falls back to the default connection.
      *
-     * @param key The key to route
-     * @return Connection to use for this key (never null)
+     * Exclusion-aware (Step 11b): nodes in @p ctx.excludeNodes are skipped, so a
+     * user-driven retry avoids the nodes a prior attempt already tried. The
+     * ordered candidate list (owners first, then non-owner proxy fallback, minus
+     * excluded nodes) is computed by the pure orderKeyCandidates() helper.
+     *
+     * @param key      The key to route
+     * @param ctx      Retry/exclusion context (defaulted: no exclusions)
+     * @param triedOut If non-null, receives every node this call attempted, in
+     *                 order — for accumulating triedNodes across retries.
+     * @return Connection to use for this key (never null). Returned as a
+     *         shared_ptr so it stays alive for the duration of the operation even
+     *         if a concurrent topology update evicts it from the pool. Acquires
+     *         stateMutex_.
+     * @throws HotRodClientException (BeforeSend) if no candidate was reachable;
+     *         ownersExhausted is set iff every owner was tried or excluded.
+     */
+    std::shared_ptr<MultiplexedConnection> selectServerForKey(const ByteArray& key,
+                                              const RetryContext& ctx = {},
+                                              std::vector<ServerAddress>* triedOut = nullptr);
+
+    /**
+     * Select a connection for a keyless operation (e.g. ping), with the same
+     * before-send failover as selectServerForKey but over *all* servers rather
+     * than a key's owners (Step 11c). Candidates are every server in the current
+     * topology minus ctx.excludeNodes, in topology order (computed by
+     * orderKeyCandidates with an empty owner list); the first reachable one wins.
+     *
+     * Ping is often one of the first calls, before any topology has been received:
+     * when the topology has no servers this returns the seed connection_ so an
+     * early ping still works. Once real topology exists, exhausting it (all
+     * servers excluded or unreachable) throws instead of silently reusing an
+     * excluded seed.
+     *
+     * @param ctx      Retry/exclusion context (defaulted: no exclusions)
+     * @param triedOut If non-null, receives every node this call attempted.
+     * @return Connection to use (never null). shared_ptr for the same lifetime
+     *         reason as selectServerForKey. Acquires stateMutex_.
+     * @throws HotRodClientException (BeforeSend) if topology exists but no
+     *         candidate was reachable. ownersExhausted is always false (keyless).
      */
-    MultiplexedConnection* selectServerForKey(const ByteArray& key);
+    std::shared_ptr<MultiplexedConnection> selectAnyServer(const RetryContext& ctx = {},
+                                              std::vector<ServerAddress>* triedOut = nullptr);
 
     /**
      * Get or create connection for a specific server.
      *
+     * Caller must hold stateMutex_ (invoked from selectServerForKey); this method
+     * does not lock.
+     *
      * @param server Server information (host, port, hashId)
      * @return MultiplexedConnection to the server (never null)
      */
-    MultiplexedConnection* getConnectionForServer(const ServerInfo& server);
+    std::shared_ptr<MultiplexedConnection> getConnectionForServer(const ServerInfo& server);
 
     /**
      * Clean up connections to servers no longer in topology.
-     * Called after topology updates.
+     * Called after topology updates. Caller must hold stateMutex_
+     * (invoked from handleTopologyUpdate); this method does not lock.
      */
     void cleanupStaleConnections();
 };
 
 } // namespace hotrod
+
+// RetryView completes the excluding()/RetryView pairing; included last so
+// RemoteCache is a complete type. (Include-guarded, so the mutual include of
+// RemoteCache.h from RetryView.h is a no-op here.)
+#include "RetryView.h"
diff --git a/include/hotrod/RetryContext.h b/include/hotrod/RetryContext.h
new file mode 100644
index 0000000..579731c
--- /dev/null
+++ b/include/hotrod/RetryContext.h
@@ -0,0 +1,35 @@
+#pragma once
+
+#include "HotRodClientException.h"  // ServerAddress
+#include <vector>
+
+namespace hotrod {
+
+/**
+ * User→library input that drives an exclusion-aware retry of an operation
+ * (Step 11b). Defaulted on every operation, so existing call sites are
+ * untouched; a retrying caller builds one from a caught HotRodClientException.
+ *
+ * Note this is deliberately *not* where proxy-to-non-owner lives: that is
+ * client-level routing policy (see RemoteCache, D5), not per-call input.
+ *
+ * Typical loop:
+ *   RetryContext ctx;
+ *   for (;;) {
+ *       try { return cache.get(key, ctx).get(); }
+ *       catch (const HotRodClientException& e) {
+ *           if (!isTransient(e)) throw;
+ *           // caller decides, from e.phase / outcomeUncertain(e), whether THIS
+ *           // op is safe to retry, then avoids the nodes already tried:
+ *           ctx.excludeNodes = e.triedNodes;
+ *       }
+ *   }
+ *
+ * See docs/ERROR_HANDLING_DESIGN.md §3.2.
+ */
+struct RetryContext {
+    std::vector<ServerAddress> excludeNodes;   // don't dispatch here again
+    // (room to grow: deadline, attempt count, ...)
+};
+
+} // namespace hotrod
diff --git a/include/hotrod/RetryView.h b/include/hotrod/RetryView.h
new file mode 100644
index 0000000..d20cc01
--- /dev/null
+++ b/include/hotrod/RetryView.h
@@ -0,0 +1,91 @@
+#pragma once
+
+#include "RemoteCache.h"
+#include "RetryContext.h"
+#include "Types.h"
+#include <future>
+#include <optional>
+#include <cstdint>
+#include <utility>
+
+namespace hotrod {
+
+/**
+ * A lightweight, non-owning view over a RemoteCache that carries a RetryContext
+ * (currently an exclusion set). Returned by RemoteCache::excluding(); it exposes
+ * the same key-routed operations, each dispatched avoiding the excluded nodes.
+ *
+ * Base RemoteCache operations stay retry-free — a caller opts into exclusion
+ * only on the retry path, seeding it from a caught exception:
+ *
+ *   try { return cache.get(key).get(); }
+ *   catch (const HotRodClientException& e) {
+ *       if (!isTransient(e)) throw;                  // + your idempotency call
+ *       return cache.excluding(e).get(key).get();    // avoids e.triedNodes
+ *   }
+ *
+ * Non-owning: a RetryView must not outlive the RemoteCache it views (the same
+ * lifetime rule as calling the cache directly). It holds no connection or other
+ * client internals — only a pointer to the cache and a copied context — so it is
+ * cheap to create per attempt.
+ *
+ * See docs/ERROR_HANDLING_DESIGN.md §3.2 / D3.
+ */
+class RetryView {
+public:
+    RetryView(RemoteCache& cache, RetryContext ctx)
+        : cache_(&cache), ctx_(std::move(ctx)) {}
+
+    std::future<void> ping() {
+        return cache_->pingImpl(ctx_);
+    }
+
+    std::future<std::optional<ByteArray>> get(const ByteArray& key) {
+        return cache_->getImpl(key, ctx_);
+    }
+
+    std::future<std::optional<EntryWithMetadata>> getWithMetadata(const ByteArray& key) {
+        return cache_->getWithMetadataImpl(key, ctx_);
+    }
+
+    std::future<std::optional<EntryWithMetadata>> put(const ByteArray& key, const ByteArray& value,
+                                                      uint64_t lifespan = 0, uint64_t maxIdle = 0,
+                                                      bool previousValue = false) {
+        return cache_->putImpl(key, value, lifespan, maxIdle, previousValue, ctx_);
+    }
+
+    std::future<std::optional<EntryWithMetadata>> putIfAbsent(const ByteArray& key, const ByteArray& value,
+                                                              uint64_t lifespan = 0, uint64_t maxIdle = 0,
+                                                              bool previousValue = false) {
+        return cache_->putIfAbsentImpl(key, value, lifespan, maxIdle, previousValue, ctx_);
+    }
+
+    std::future<std::optional<EntryWithMetadata>> replace(const ByteArray& key, const ByteArray& value,
+                                                          uint64_t lifespan = 0, uint64_t maxIdle = 0,
+                                                          bool previousValue = false) {
+        return cache_->replaceImpl(key, value, lifespan, maxIdle, previousValue, ctx_);
+    }
+
+    std::future<bool> containsKey(const ByteArray& key) {
+        return cache_->containsKeyImpl(key, ctx_);
+    }
+
+    std::future<std::optional<EntryWithMetadata>> remove(const ByteArray& key, bool previousValue = false) {
+        return cache_->removeImpl(key, previousValue, ctx_);
+    }
+
+    std::future<bool> removeWithVersion(const ByteArray& key, int64_t version) {
+        return cache_->removeWithVersionImpl(key, version, ctx_);
+    }
+
+    std::future<bool> replaceWithVersion(const ByteArray& key, const ByteArray& value,
+                                         int64_t version, uint64_t lifespan = 0, uint64_t maxIdle = 0) {
+        return cache_->replaceWithVersionImpl(key, value, version, lifespan, maxIdle, ctx_);
+    }
+
+private:
+    RemoteCache* cache_;   // non-owning; must outlive this view
+    RetryContext ctx_;     // bound exclusion set
+};
+
+} // namespace hotrod
diff --git a/include/hotrod/ServerSelection.h b/include/hotrod/ServerSelection.h
new file mode 100644
index 0000000..69e692c
--- /dev/null
+++ b/include/hotrod/ServerSelection.h
@@ -0,0 +1,75 @@
+#pragma once
+
+#include "HotRodClientException.h"  // ServerAddress
+#include <vector>
+#include <algorithm>
+
+namespace hotrod {
+
+/**
+ * Build the ordered list of servers to try for a key.
+ *
+ * Owners come first, in owner order (primary, then backups). When
+ * @p proxyToNonOwner is true (the default, matching Java), the remaining servers
+ * follow as a non-owner *proxy* fallback; when false, only owners are returned
+ * so a caller that exhausts them can report ownersExhausted rather than silently
+ * proxying (D5). Nodes in @p excludeNodes are dropped, and duplicates are
+ * collapsed (a server that is both an owner and present in @p allServers appears
+ * once, keeping its owner position).
+ *
+ * This is pure — no I/O, no client state — so the routing decision can be
+ * unit-tested in isolation from connection acquisition. Connection acquisition
+ * (and its failures) is the caller's job; see RemoteCache::selectServerForKey.
+ *
+ * See docs/ERROR_HANDLING_DESIGN.md §3.2 / D5 (Step 11b).
+ *
+ * @param owners      Owners for the key's segment, in preference order.
+ * @param allServers  All servers currently in the topology.
+ * @param excludeNodes Nodes to skip (e.g. already tried on a prior attempt).
+ * @param proxyToNonOwner If true, append non-owner servers as proxy fallback.
+ * @return Ordered, de-duplicated candidate list with excluded nodes removed.
+ */
+inline std::vector<ServerAddress> orderKeyCandidates(
+    const std::vector<ServerAddress>& owners,
+    const std::vector<ServerAddress>& allServers,
+    const std::vector<ServerAddress>& excludeNodes,
+    bool proxyToNonOwner = true)
+{
+    std::vector<ServerAddress> out;
+    out.reserve(owners.size() + allServers.size());
+
+    auto isExcluded = [&](const ServerAddress& s) {
+        return std::find(excludeNodes.begin(), excludeNodes.end(), s) != excludeNodes.end();
+    };
+    auto pushUnique = [&](const ServerAddress& s) {
+        if (isExcluded(s)) return;
+        if (std::find(out.begin(), out.end(), s) != out.end()) return;
+        out.push_back(s);
+    };
+
+    for (const auto& o : owners) pushUnique(o);        // owners first, in order
+    if (proxyToNonOwner)
+        for (const auto& s : allServers) pushUnique(s);  // then non-owner fallback
+    return out;
+}
+
+/**
+ * Set-union of two node lists: the members of @p a in order, followed by the
+ * members of @p b not already present. Used to accumulate an operation's
+ * triedNodes across successive retries — the exclusion set carried in from prior
+ * attempts, unioned with the nodes the current attempt touched — so feeding a
+ * caught exception's triedNodes back as the next attempt's exclusion set grows
+ * monotonically without duplicates (Step 11b).
+ *
+ * See docs/ERROR_HANDLING_DESIGN.md §3.1 (triedNodes union across retries).
+ */
+inline std::vector<ServerAddress> unionNodes(std::vector<ServerAddress> a,
+                                             const std::vector<ServerAddress>& b)
+{
+    for (const auto& n : b)
+        if (std::find(a.begin(), a.end(), n) == a.end())
+            a.push_back(n);
+    return a;
+}
+
+} // namespace hotrod
diff --git a/src/operations/RemoteCache.cpp b/src/operations/RemoteCache.cpp
index efe032f..86fdbea 100644
--- a/src/operations/RemoteCache.cpp
+++ b/src/operations/RemoteCache.cpp
@@ -1,11 +1,35 @@
 #include "hotrod/RemoteCache.h"
+#include "hotrod/RetryView.h"
 #include "hotrod/Codec.h"
 #include "hotrod/HotRodClientException.h"
+#include "hotrod/ServerSelection.h"
 #include <stdexcept>
 #include <set>
+#include <algorithm>
 
 namespace hotrod
 {
+   namespace
+   {
+      // Rethrow a failed Response's error, enriching a HotRodClientException with
+      // the union of nodes tried across this and prior attempts so a caller can
+      // feed e.triedNodes back as the next attempt's exclusion set. Non-HotRod
+      // exceptions propagate unchanged. Always throws.
+      [[noreturn]] void rethrowWithTriedNodes(std::exception_ptr eptr,
+                                              const std::vector<ServerAddress> &tried,
+                                              const std::vector<ServerAddress> &excluded)
+      {
+         try
+         {
+            std::rethrow_exception(eptr);
+         }
+         catch (HotRodClientException &e)
+         {
+            e.triedNodes = unionNodes(unionNodes(excluded, tried), e.triedNodes);
+            throw;
+         }
+      }
+   } // namespace
 
    RemoteCache::RemoteCache(const std::string &host, uint16_t port)
        : host_(host), port_(port), cacheName_(""),
@@ -17,7 +41,7 @@ namespace hotrod
       {
          this->handleTopologyUpdate(topo, cacheName);
       };
-      connection_ = std::make_unique<MultiplexedConnection>(host, port, topologyCallback);
+      connection_ = std::make_shared<MultiplexedConnection>(host, port, topologyCallback);
    }
 
    RemoteCache::RemoteCache(const std::string &host, uint16_t port, const std::string &cacheName)
@@ -30,7 +54,7 @@ namespace hotrod
       {
          this->handleTopologyUpdate(topo, cacheName);
       };
-      connection_ = std::make_unique<MultiplexedConnection>(host, port, topologyCallback);
+      connection_ = std::make_shared<MultiplexedConnection>(host, port, topologyCallback);
    }
 
    RemoteCache::~RemoteCache()
@@ -60,6 +84,11 @@ namespace hotrod
 
    void RemoteCache::handleTopologyUpdate(const TopologyInfo &topo, const std::string & /*cacheName*/)
    {
+      // Runs on a connection's read-loop thread. Guard the routing/pool state
+      // against user/retry threads in selectServerForKey (slice 4). cleanup below
+      // runs with the lock held (it must not re-lock).
+      std::lock_guard<std::mutex> lock(stateMutex_);
+
       // Update topology
       topology_ = topo;
 
@@ -74,6 +103,11 @@ namespace hotrod
    }
 
    std::future<void> RemoteCache::ping()
+   {
+      return pingImpl(RetryContext{});
+   }
+
+   std::future<void> RemoteCache::pingImpl(const RetryContext &ctx)
    {
       // Build request body (empty for PING)
       ByteArray requestBody;
@@ -206,21 +240,33 @@ namespace hotrod
          return {}; // PING doesn't return data
       };
 
-      // Use default connection (PING is not key-based)
-      auto responseFuture = connection_->execute(requestBody, 0x17, 0x18, bodyParser, cacheName_);
+      // PING is keyless: select any reachable server (auto before-send failover
+      // across the whole cluster, minus ctx.excludeNodes), recording tried nodes.
+      std::vector<ServerAddress> tried;
+      std::shared_ptr<MultiplexedConnection> conn = selectAnyServer(ctx, &tried);
+
+      auto responseFuture = conn->execute(requestBody, 0x17, 0x18, bodyParser, cacheName_);
 
-      // Transform Response → void
+      // Transform Response → void. On failure enrich the exception with the union
+      // of excluded + tried nodes so a caller can feed it back via excluding(e).
       return std::async(std::launch::deferred,
-                        [responseFuture = std::move(responseFuture)]() mutable
+                        [conn, responseFuture = std::move(responseFuture),
+                         tried = std::move(tried),
+                         excluded = ctx.excludeNodes]() mutable
                         {
                            Response resp = responseFuture.get();
                            if (resp.error)
-                              std::rethrow_exception(resp.error);
+                              rethrowWithTriedNodes(resp.error, tried, excluded);
                            // Success - return void
                         });
    }
 
    std::future<std::optional<ByteArray>> RemoteCache::get(const ByteArray &key)
+   {
+      return getImpl(key, RetryContext{});
+   }
+
+   std::future<std::optional<ByteArray>> RemoteCache::getImpl(const ByteArray &key, const RetryContext &ctx)
    {
       // Build request body (just the key)
       ByteArray requestBody;
@@ -244,8 +290,9 @@ namespace hotrod
          return conn->receiveByteArray();
       };
 
-      // Select connection (hash-aware routing or default)
-      MultiplexedConnection *conn = selectServerForKey(key);
+      // Select connection (hash-aware routing or default), recording tried nodes
+      std::vector<ServerAddress> tried;
+      std::shared_ptr<MultiplexedConnection> conn = selectServerForKey(key, ctx, &tried);
 
       // Execute and get Response future
       auto responseFuture = conn->execute(
@@ -257,13 +304,15 @@ namespace hotrod
 
       // Transform Response → std::optional<ByteArray>
       return std::async(std::launch::deferred,
-                        [responseFuture = std::move(responseFuture)]() mutable -> std::optional<ByteArray>
+                        [conn, responseFuture = std::move(responseFuture),
+                         tried = std::move(tried),
+                         excluded = ctx.excludeNodes]() mutable -> std::optional<ByteArray>
                         {
                            Response resp = responseFuture.get();
 
                            if (resp.error)
                            {
-                              std::rethrow_exception(resp.error);
+                              rethrowWithTriedNodes(resp.error, tried, excluded);
                            }
 
                            if (resp.status == 0x00 && resp.body.has_value())
@@ -276,6 +325,11 @@ namespace hotrod
    }
 
    std::future<std::optional<EntryWithMetadata>> RemoteCache::getWithMetadata(const ByteArray &key)
+   {
+      return getWithMetadataImpl(key, RetryContext{});
+   }
+
+   std::future<std::optional<EntryWithMetadata>> RemoteCache::getWithMetadataImpl(const ByteArray &key, const RetryContext &ctx)
    {
       // Build request body (just the key) - identical to GET
       ByteArray requestBody;
@@ -303,8 +357,9 @@ namespace hotrod
          return entry;
       };
 
-      // Select connection (hash-aware routing or default)
-      MultiplexedConnection *conn = selectServerForKey(key);
+      // Select connection (hash-aware routing or default), recording tried nodes
+      std::vector<ServerAddress> tried;
+      std::shared_ptr<MultiplexedConnection> conn = selectServerForKey(key, ctx, &tried);
 
       // Execute and get Response future
       auto responseFuture = conn->execute(
@@ -316,13 +371,15 @@ namespace hotrod
 
       // Transform Response → std::optional<EntryWithMetadata>
       return std::async(std::launch::deferred,
-                        [responseFuture = std::move(responseFuture)]() mutable -> std::optional<EntryWithMetadata>
+                        [conn, responseFuture = std::move(responseFuture),
+                         tried = std::move(tried),
+                         excluded = ctx.excludeNodes]() mutable -> std::optional<EntryWithMetadata>
                         {
                            Response resp = responseFuture.get();
 
                            if (resp.error)
                            {
-                              std::rethrow_exception(resp.error);
+                              rethrowWithTriedNodes(resp.error, tried, excluded);
                            }
 
                            if (resp.status == 0x00 && resp.body.has_value())
@@ -336,6 +393,13 @@ namespace hotrod
 
    std::future<std::optional<EntryWithMetadata>> RemoteCache::put(const ByteArray &key, const ByteArray &value,
                                                           uint64_t lifespan, uint64_t maxIdle, bool previousValue)
+   {
+      return putImpl(key, value, lifespan, maxIdle, previousValue, RetryContext{});
+   }
+
+   std::future<std::optional<EntryWithMetadata>> RemoteCache::putImpl(const ByteArray &key, const ByteArray &value,
+                                                          uint64_t lifespan, uint64_t maxIdle, bool previousValue,
+                                                          const RetryContext &ctx)
    {
       // Build PUT request body
       // Encode request header + key + expiration + value
@@ -402,18 +466,22 @@ namespace hotrod
          return {};
       };
       // Execute
-      MultiplexedConnection *conn = selectServerForKey(key);
+      std::vector<ServerAddress> tried;
+      std::shared_ptr<MultiplexedConnection> conn = selectServerForKey(key, ctx, &tried);
       auto responseFuture = conn->execute(requestBody, 0x01, 0x02, bodyParser, cacheName_, (previousValue ? 1 : 0));
       // Transform Response → std::optional<ByteArray>
       return std::async(std::launch::deferred,
-                        [responseFuture = std::move(responseFuture)]() mutable -> std::optional<EntryWithMetadata>
+                        [conn, responseFuture = std::move(responseFuture),
+                         tried = std::move(tried),
+                         excluded = ctx.excludeNodes]() mutable -> std::optional<EntryWithMetadata>
                         {
                            Response resp = responseFuture.get();
                            if (resp.error)
-                              std::rethrow_exception(resp.error);
+                              rethrowWithTriedNodes(resp.error, tried, excluded);
                            if (resp.status != 0x00 && resp.status != 0x03)
                               throw HotRodClientException("PUT failed with status: " + std::to_string(resp.status),
-                                                         FailurePhase::ServerError, resp.status);
+                                                         FailurePhase::ServerError, resp.status,
+                                                         unionNodes(excluded, tried));
                            if (resp.status == 0x03 && resp.body.has_value()) {
                               return std::any_cast<EntryWithMetadata>(resp.body);
                            }
@@ -422,6 +490,12 @@ namespace hotrod
    }
 
    std::future<std::optional<EntryWithMetadata>> RemoteCache::remove(const ByteArray &key, bool previousValue)
+   {
+      return removeImpl(key, previousValue, RetryContext{});
+   }
+
+   std::future<std::optional<EntryWithMetadata>> RemoteCache::removeImpl(const ByteArray &key, bool previousValue,
+                                                          const RetryContext &ctx)
    {
       // Build REMOVE request body
       ByteArray requestBody;
@@ -456,16 +530,19 @@ namespace hotrod
       };
 
       // Execute
-      MultiplexedConnection *conn = selectServerForKey(key);
+      std::vector<ServerAddress> tried;
+      std::shared_ptr<MultiplexedConnection> conn = selectServerForKey(key, ctx, &tried);
       auto responseFuture = conn->execute(requestBody, 0x0B, 0x0C, bodyParser, cacheName_, (previousValue ? 1 : 0));
 
       // Transform Response → std::optional<ByteArray>
       return std::async(std::launch::deferred,
-                        [responseFuture = std::move(responseFuture)]() mutable -> std::optional<EntryWithMetadata>
+                        [conn, responseFuture = std::move(responseFuture),
+                         tried = std::move(tried),
+                         excluded = ctx.excludeNodes]() mutable -> std::optional<EntryWithMetadata>
                         {
                            Response resp = responseFuture.get();
                            if (resp.error)
-                              std::rethrow_exception(resp.error);
+                              rethrowWithTriedNodes(resp.error, tried, excluded);
                            if (resp.status == 0x03 && resp.body.has_value())
                               return std::any_cast<EntryWithMetadata>(resp.body);
                            return std::nullopt;
@@ -473,6 +550,12 @@ namespace hotrod
    }
 
    std::future<bool> RemoteCache::removeWithVersion(const ByteArray &key, int64_t version)
+   {
+      return removeWithVersionImpl(key, version, RetryContext{});
+   }
+
+   std::future<bool> RemoteCache::removeWithVersionImpl(const ByteArray &key, int64_t version,
+                                                        const RetryContext &ctx)
    {
       // Build REMOVE_WITH_VERSION request body: key (lp_bytes) + version (8-byte
       // big-endian s8). Mirrors Java RemoveIfUnmodifiedOperation which appends
@@ -508,23 +591,33 @@ namespace hotrod
       };
 
       // Execute
-      MultiplexedConnection *conn = selectServerForKey(key);
+      std::vector<ServerAddress> tried;
+      std::shared_ptr<MultiplexedConnection> conn = selectServerForKey(key, ctx, &tried);
       auto responseFuture = conn->execute(requestBody, 0x0D, 0x0E, bodyParser, cacheName_);
 
       // Transform Response → bool (removed?). Success is 0x00 or 0x03, matching
       // Java's HotRodConstants.isSuccess / RspCode.isUpdated().
       return std::async(std::launch::deferred,
-                        [responseFuture = std::move(responseFuture)]() mutable -> bool
+                        [conn, responseFuture = std::move(responseFuture),
+                         tried = std::move(tried),
+                         excluded = ctx.excludeNodes]() mutable -> bool
                         {
                            Response resp = responseFuture.get();
                            if (resp.error)
-                              std::rethrow_exception(resp.error);
+                              rethrowWithTriedNodes(resp.error, tried, excluded);
                            return resp.status == 0x00 || resp.status == 0x03;
                         });
    }
 
    std::future<bool> RemoteCache::replaceWithVersion(const ByteArray &key, const ByteArray &value,
                                                      int64_t version, uint64_t lifespan, uint64_t maxIdle)
+   {
+      return replaceWithVersionImpl(key, value, version, lifespan, maxIdle, RetryContext{});
+   }
+
+   std::future<bool> RemoteCache::replaceWithVersionImpl(const ByteArray &key, const ByteArray &value,
+                                                     int64_t version, uint64_t lifespan, uint64_t maxIdle,
+                                                     const RetryContext &ctx)
    {
       // Build REPLACE_WITH_VERSION request body:
       //   key (lp_bytes) + time_units (u1) + [lifespan vLong] + [maxIdle vLong]
@@ -580,23 +673,33 @@ namespace hotrod
       };
 
       // Execute
-      MultiplexedConnection *conn = selectServerForKey(key);
+      std::vector<ServerAddress> tried;
+      std::shared_ptr<MultiplexedConnection> conn = selectServerForKey(key, ctx, &tried);
       auto responseFuture = conn->execute(requestBody, 0x09, 0x0A, bodyParser, cacheName_);
 
       // Transform Response → bool (replaced?). Success is 0x00 or 0x03, matching
       // Java's HotRodConstants.isSuccess / RspCode.isUpdated().
       return std::async(std::launch::deferred,
-                        [responseFuture = std::move(responseFuture)]() mutable -> bool
+                        [conn, responseFuture = std::move(responseFuture),
+                         tried = std::move(tried),
+                         excluded = ctx.excludeNodes]() mutable -> bool
                         {
                            Response resp = responseFuture.get();
                            if (resp.error)
-                              std::rethrow_exception(resp.error);
+                              rethrowWithTriedNodes(resp.error, tried, excluded);
                            return resp.status == 0x00 || resp.status == 0x03;
                         });
    }
 
    std::future<std::optional<EntryWithMetadata>> RemoteCache::putIfAbsent(const ByteArray &key, const ByteArray &value,
                                                                           uint64_t lifespan, uint64_t maxIdle, bool previousValue)
+   {
+      return putIfAbsentImpl(key, value, lifespan, maxIdle, previousValue, RetryContext{});
+   }
+
+   std::future<std::optional<EntryWithMetadata>> RemoteCache::putIfAbsentImpl(const ByteArray &key, const ByteArray &value,
+                                                                          uint64_t lifespan, uint64_t maxIdle, bool previousValue,
+                                                                          const RetryContext &ctx)
    {
       // Build PUT_IF_ABSENT request body: identical layout to PUT
       //   key (lp_bytes) + time_units (u1) + [lifespan vLong] + [maxIdle vLong]
@@ -641,17 +744,20 @@ namespace hotrod
                                      FailurePhase::ServerError, status);
       };
 
-      MultiplexedConnection *conn = selectServerForKey(key);
+      std::vector<ServerAddress> tried;
+      std::shared_ptr<MultiplexedConnection> conn = selectServerForKey(key, ctx, &tried);
       auto responseFuture = conn->execute(requestBody, 0x05, 0x06, bodyParser, cacheName_, (previousValue ? 1 : 0));
 
       // Transform Response → optional<EntryWithMetadata>: the existing entry that
       // blocked storage, if returned; nullopt when stored (or no body present).
       return std::async(std::launch::deferred,
-                        [responseFuture = std::move(responseFuture)]() mutable -> std::optional<EntryWithMetadata>
+                        [conn, responseFuture = std::move(responseFuture),
+                         tried = std::move(tried),
+                         excluded = ctx.excludeNodes]() mutable -> std::optional<EntryWithMetadata>
                         {
                            Response resp = responseFuture.get();
                            if (resp.error)
-                              std::rethrow_exception(resp.error);
+                              rethrowWithTriedNodes(resp.error, tried, excluded);
                            if ((resp.status == 0x03 || resp.status == 0x04) && resp.body.has_value())
                               return std::any_cast<EntryWithMetadata>(resp.body);
                            return std::nullopt;
@@ -660,6 +766,13 @@ namespace hotrod
 
    std::future<std::optional<EntryWithMetadata>> RemoteCache::replace(const ByteArray &key, const ByteArray &value,
                                                                       uint64_t lifespan, uint64_t maxIdle, bool previousValue)
+   {
+      return replaceImpl(key, value, lifespan, maxIdle, previousValue, RetryContext{});
+   }
+
+   std::future<std::optional<EntryWithMetadata>> RemoteCache::replaceImpl(const ByteArray &key, const ByteArray &value,
+                                                                      uint64_t lifespan, uint64_t maxIdle, bool previousValue,
+                                                                      const RetryContext &ctx)
    {
       // Build REPLACE request body: identical layout to PUT
       //   key (lp_bytes) + time_units (u1) + [lifespan vLong] + [maxIdle vLong]
@@ -702,17 +815,20 @@ namespace hotrod
                                      FailurePhase::ServerError, status);
       };
 
-      MultiplexedConnection *conn = selectServerForKey(key);
+      std::vector<ServerAddress> tried;
+      std::shared_ptr<MultiplexedConnection> conn = selectServerForKey(key, ctx, &tried);
       auto responseFuture = conn->execute(requestBody, 0x07, 0x08, bodyParser, cacheName_, (previousValue ? 1 : 0));
 
       // Transform Response → optional<EntryWithMetadata>: the replaced (previous)
       // entry, if returned; nullopt when nothing was replaced (or no body).
       return std::async(std::launch::deferred,
-                        [responseFuture = std::move(responseFuture)]() mutable -> std::optional<EntryWithMetadata>
+                        [conn, responseFuture = std::move(responseFuture),
+                         tried = std::move(tried),
+                         excluded = ctx.excludeNodes]() mutable -> std::optional<EntryWithMetadata>
                         {
                            Response resp = responseFuture.get();
                            if (resp.error)
-                              std::rethrow_exception(resp.error);
+                              rethrowWithTriedNodes(resp.error, tried, excluded);
                            if ((resp.status == 0x03 || resp.status == 0x04) && resp.body.has_value())
                               return std::any_cast<EntryWithMetadata>(resp.body);
                            return std::nullopt;
@@ -720,6 +836,11 @@ namespace hotrod
    }
 
    std::future<bool> RemoteCache::containsKey(const ByteArray &key)
+   {
+      return containsKeyImpl(key, RetryContext{});
+   }
+
+   std::future<bool> RemoteCache::containsKeyImpl(const ByteArray &key, const RetryContext &ctx)
    {
       // Build CONTAINS_KEY request body (just the key), like GET.
       ByteArray requestBody;
@@ -740,22 +861,45 @@ namespace hotrod
                                      FailurePhase::ServerError, status);
       };
 
-      MultiplexedConnection *conn = selectServerForKey(key);
+      std::vector<ServerAddress> tried;
+      std::shared_ptr<MultiplexedConnection> conn = selectServerForKey(key, ctx, &tried);
       auto responseFuture = conn->execute(requestBody, 0x0F, 0x10, bodyParser, cacheName_);
 
       // Transform Response → bool (key present?). Success is 0x00.
       return std::async(std::launch::deferred,
-                        [responseFuture = std::move(responseFuture)]() mutable -> bool
+                        [conn, responseFuture = std::move(responseFuture),
+                         tried = std::move(tried),
+                         excluded = ctx.excludeNodes]() mutable -> bool
                         {
                            Response resp = responseFuture.get();
                            if (resp.error)
-                              std::rethrow_exception(resp.error);
+                              rethrowWithTriedNodes(resp.error, tried, excluded);
                            return resp.status == 0x00;
                         });
    }
 
-   MultiplexedConnection *RemoteCache::selectServerForKey(const ByteArray &key)
+   RetryView RemoteCache::excluding(std::vector<ServerAddress> excludeNodes)
    {
+      return RetryView(*this, RetryContext{std::move(excludeNodes)});
+   }
+
+   RetryView RemoteCache::excluding(const HotRodClientException &e)
+   {
+      return RetryView(*this, RetryContext{e.triedNodes});
+   }
+
+   std::shared_ptr<MultiplexedConnection> RemoteCache::selectServerForKey(const ByteArray &key,
+                                                          const RetryContext &ctx,
+                                                          std::vector<ServerAddress> *triedOut)
+   {
+      // Guard the routing/pool state against concurrent topology callbacks on the
+      // read-loop thread (handleTopologyUpdate). Held across connection
+      // acquisition (getConnectionForServer, which may connect()); the caller runs
+      // the actual request on the returned connection *after* the lock is
+      // released. The shared_ptr keeps that connection alive even if a topology
+      // update evicts it from the pool meanwhile (slice 4).
+      std::lock_guard<std::mutex> lock(stateMutex_);
+
       // If hash-aware routing is enabled and hash topology is available
       if (clientIntelligence_ == ClientIntelligence::HASH_DISTRIBUTION_AWARE &&
           consistentHash_.hasHashTopology())
@@ -768,61 +912,71 @@ namespace hotrod
 
          if (!owners.empty())
          {
-            // Try each owner in order (primary first, then backups)
-            for (size_t i = 0; i < owners.size(); i++)
+            // Build the ordered candidate list: owners first, then (when
+            // proxyToNonOwner_ is set, D5) non-owner proxy fallback, minus any
+            // excluded (already-tried) nodes. This routing decision is pure and
+            // unit-tested (orderKeyCandidates); only the connection acquisition
+            // below does I/O.
+            std::vector<ServerAddress> ownerAddrs;
+            ownerAddrs.reserve(owners.size());
+            for (const auto *owner : owners)
+               ownerAddrs.push_back({owner->host, owner->port});
+
+            std::vector<ServerAddress> allAddrs;
+            for (const auto &server : topology_.getServers())
+               allAddrs.push_back({server.host, server.port});
+
+            std::vector<ServerAddress> candidates =
+                orderKeyCandidates(ownerAddrs, allAddrs, ctx.excludeNodes,
+                                   proxyToNonOwner_);
+
+            // Try each candidate in order, recording every node we attempt so
+            // the caller can accumulate triedNodes across retries.
+            std::vector<ServerAddress> tried;
+            for (const auto &cand : candidates)
             {
-               const ServerInfo *owner = owners[i];
+               tried.push_back(cand);
                try
                {
-                  MultiplexedConnection *conn = getConnectionForServer(*owner);
+                  // hashId is unused by getConnectionForServer (host:port only).
+                  std::shared_ptr<MultiplexedConnection> conn =
+                      getConnectionForServer(ServerInfo(cand.host, cand.port, 0));
+                  if (triedOut)
+                     *triedOut = tried;
                   return conn;
                }
                catch (const std::exception &e)
                {
-                  fprintf(stderr, "[WARN] Failed to connect to owner %s:%u: %s\n",
-                          owner->host.c_str(), owner->port, e.what());
-                  // Continue to next owner
+                  fprintf(stderr, "[WARN] Failed to connect to %s:%u: %s\n",
+                          cand.host.c_str(), cand.port, e.what());
+                  // Continue to next candidate
                }
             }
 
-            // All owners failed, try any server in topology
-            fprintf(stderr, "[WARN] All owners failed for segment %d, trying any server\n", segment);
-            const auto &servers = topology_.getServers();
-            for (const auto &server : servers)
-            {
-               // Skip servers we already tried as owners
-               bool alreadyTried = false;
-               for (const auto *owner : owners)
-               {
-                  if (server.host == owner->host && server.port == owner->port)
-                  {
-                     alreadyTried = true;
-                     break;
-                  }
-               }
-               if (alreadyTried)
-                  continue;
-
-               try
-               {
-                  MultiplexedConnection *conn = getConnectionForServer(server);
-                  return conn;
-               }
-               catch (const std::exception &e)
+            // No candidate was reachable. Nothing was sent (BeforeSend). An
+            // owner is "exhausted" if it was tried on this attempt or excluded
+            // (tried on a prior attempt); ownersExhausted holds iff that covers
+            // every owner — i.e. further retry can only reach a proxy.
+            auto attempted = [&](const ServerAddress &a) {
+               return std::find(tried.begin(), tried.end(), a) != tried.end() ||
+                      std::find(ctx.excludeNodes.begin(), ctx.excludeNodes.end(), a) !=
+                          ctx.excludeNodes.end();
+            };
+            bool ownersExhausted = true;
+            for (const auto &o : ownerAddrs)
+               if (!attempted(o))
                {
-                  fprintf(stderr, "[WARN] Failed to connect to server %s:%u: %s\n",
-                          server.host.c_str(), server.port, e.what());
-                  // Continue to next server
+                  ownersExhausted = false;
+                  break;
                }
-            }
 
-            // No servers available: every owner and fallback server was
-            // unreachable. Nothing was sent (BeforeSend); ownersExhausted marks
-            // that all owners were tried. (The retry loop that treats exhaustion
-            // as terminal/non-transient lands in 11b.)
+            fprintf(stderr, "[WARN] No servers available for segment %d\n", segment);
+            if (triedOut)
+               *triedOut = tried;
             throw HotRodClientException("No servers available in topology",
                                         FailurePhase::BeforeSend, std::nullopt,
-                                        {}, /*ownersExhausted=*/true);
+                                        unionNodes(ctx.excludeNodes, tried),
+                                        ownersExhausted);
          }
          else
          {
@@ -831,10 +985,68 @@ namespace hotrod
       }
 
       // Fallback: use default connection
-      return connection_.get();
+      return connection_;
+   }
+
+   std::shared_ptr<MultiplexedConnection> RemoteCache::selectAnyServer(const RetryContext &ctx,
+                                                          std::vector<ServerAddress> *triedOut)
+   {
+      // Same lock/lifetime discipline as selectServerForKey (slice 4): guard the
+      // routing/pool state and keep the returned connection alive past the lock.
+      std::lock_guard<std::mutex> lock(stateMutex_);
+
+      std::vector<ServerAddress> allAddrs;
+      for (const auto &server : topology_.getServers())
+         allAddrs.push_back({server.host, server.port});
+
+      // Early ping: no topology received yet → nothing to route over, so use the
+      // seed connection so the very first ping still works.
+      if (allAddrs.empty())
+      {
+         if (triedOut)
+            triedOut->clear();
+         return connection_;
+      }
+
+      // Keyless candidates: every server minus the exclusion set, in topology
+      // order. Reuses the pure, unit-tested orderKeyCandidates() with an empty
+      // owner list (no key → no owners); the proxy flag is irrelevant here since
+      // there are no owners, so all servers come from the allServers pass.
+      std::vector<ServerAddress> candidates =
+          orderKeyCandidates({}, allAddrs, ctx.excludeNodes, true);
+
+      std::vector<ServerAddress> tried;
+      for (const auto &cand : candidates)
+      {
+         tried.push_back(cand);
+         try
+         {
+            std::shared_ptr<MultiplexedConnection> conn =
+                getConnectionForServer(ServerInfo(cand.host, cand.port, 0));
+            if (triedOut)
+               *triedOut = tried;
+            return conn;
+         }
+         catch (const std::exception &e)
+         {
+            fprintf(stderr, "[WARN] Failed to connect to %s:%u: %s\n",
+                    cand.host.c_str(), cand.port, e.what());
+            // Continue to next candidate
+         }
+      }
+
+      // Topology existed but every server was excluded or unreachable. Nothing was
+      // sent (BeforeSend). ownersExhausted is always false for a keyless op.
+      fprintf(stderr, "[WARN] No servers available for keyless operation\n");
+      if (triedOut)
+         *triedOut = tried;
+      throw HotRodClientException("No servers available in topology",
+                                  FailurePhase::BeforeSend, std::nullopt,
+                                  unionNodes(ctx.excludeNodes, tried),
+                                  /*ownersExhausted=*/false);
    }
 
-   MultiplexedConnection *RemoteCache::getConnectionForServer(const ServerInfo &server)
+   std::shared_ptr<MultiplexedConnection> RemoteCache::getConnectionForServer(const ServerInfo &server)
    {
       // Create connection pool key
       std::string poolKey = server.host + ":" + std::to_string(server.port);
@@ -846,7 +1058,7 @@ namespace hotrod
          // Connection exists, check if it's still connected
          if (it->second->isConnected())
          {
-            return it->second.get();
+            return it->second;
          }
          else
          {
@@ -863,17 +1075,16 @@ namespace hotrod
          this->handleTopologyUpdate(topo, cacheName);
       };
 
-      auto newConnection = std::make_unique<MultiplexedConnection>(server.host, server.port, topologyCallback);
+      auto newConnection = std::make_shared<MultiplexedConnection>(server.host, server.port, topologyCallback);
       newConnection->setProtocolVersion(protocolVersion_);
       newConnection->setClientIntelligence(clientIntelligence_);
 
       // connect() can throw if server is unreachable - let it propagate to caller
       newConnection->connect();
 
-      MultiplexedConnection *connPtr = newConnection.get();
-      connectionPool_[poolKey] = std::move(newConnection);
+      connectionPool_[poolKey] = newConnection;
 
-      return connPtr;
+      return newConnection;
    }
 
    void RemoteCache::cleanupStaleConnections()
diff --git a/tests/integration/PingRetryIntegrationTest.cpp b/tests/integration/PingRetryIntegrationTest.cpp
new file mode 100644
index 0000000..338c9e0
--- /dev/null
+++ b/tests/integration/PingRetryIntegrationTest.cpp
@@ -0,0 +1,192 @@
+#include <gtest/gtest.h>
+#include "TopologyTestFixture.h"
+#include "hotrod/RemoteCache.h"
+#include "hotrod/HotRodClientException.h"
+#include <thread>
+#include <chrono>
+#include <vector>
+#include <string>
+#include <algorithm>
+
+using namespace hotrod;
+using namespace hotrod::test;
+
+/**
+ * Step 11c — keyless-op retry, end-to-end (ping).
+ *
+ * ping() carries no key/owner, so it routes through selectAnyServer(): every
+ * server is a candidate in topology order (minus an exclusion set), with the
+ * same two-tier model as keyed ops — automatic before-send failover across
+ * servers, plus user-decided after-send retry via cache.excluding(e).ping().
+ * These tests assert that keyless dispatch on a live cluster:
+ *
+ *   - excluding({a live server}) still pings successfully (routes elsewhere)
+ *   - excluding(all servers) throws BeforeSend with ownersExhausted=FALSE
+ *     (the keyless distinction from keyed selectServerForKey, which sets it true)
+ *   - the catch → excluding(e).ping() loop recovers after a node is killed
+ *
+ * See docs/DECISIONS.md (2026-09-25 Step 11c entry) and
+ * docs/ERROR_HANDLING_DESIGN.md §3.2 / D2.
+ */
+class PingRetryIntegrationTest : public TopologyTest {
+protected:
+    void SetUp() override {
+        TopologyTest::SetUp();
+        resetCluster(3);
+        clearTestCache();
+    }
+
+    static ServerAddress addrOf(const ServerInfo& s) {
+        return ServerAddress{s.host, s.port};
+    }
+
+    // Ping until the client has learned the cluster topology (ping triggers a
+    // topology update in its response header, exactly like the keyed ops).
+    std::vector<ServerAddress> pingUntilTopologyKnown(RemoteCache* cache) {
+        for (int i = 0; i < 20; ++i) {
+            cache->ping().get();
+            std::this_thread::sleep_for(std::chrono::milliseconds(300));
+            const auto& servers = cache->getTopology().getServers();
+            if (servers.size() >= 2u) {
+                std::vector<ServerAddress> addrs;
+                for (const auto& s : servers) addrs.push_back(addrOf(s));
+                return addrs;
+            }
+        }
+        return {};
+    }
+
+    // Map a topology address to its cluster node number (1-based). The topology
+    // reports internal container addresses (e.g. 172.18.0.x:11222), not the
+    // published localhost:1132x ports, so we map by position: the topology is
+    // ordered node1, node2, … — the same convention RetryViewIntegrationTest uses.
+    int nodeNumberFor(RemoteCache* cache, const ServerAddress& a) {
+        const auto& topo = cache->getTopology().getServers();
+        for (size_t i = 0; i < topo.size(); ++i) {
+            if (addrOf(topo[i]) == a) return static_cast<int>(i) + 1;
+        }
+        return -1;
+    }
+};
+
+// Test 1: a plain ping() works across the cluster, and excluding one live server
+// still succeeds — selectAnyServer routes the keyless op to another node.
+// Deterministic; no node is killed.
+TEST_F(PingRetryIntegrationTest, ExcludingRoutesToAnotherServer) {
+    auto cache = createDefaultClient();
+    ASSERT_NE(nullptr, cache);
+
+    auto servers = pingUntilTopologyKnown(cache);
+    ASSERT_GE(servers.size(), 2u) << "ping must populate a multi-node topology";
+
+    // Plain keyless ping succeeds.
+    ASSERT_NO_THROW(cache->ping().get());
+
+    // Excluding the first candidate still pings — routed to a different server.
+    const ServerAddress first = servers.front();
+    fprintf(stderr, "[TEST] excluding %s:%u, ping must land elsewhere\n",
+            first.host.c_str(), first.port);
+    ASSERT_NO_THROW(cache->excluding({first}).ping().get());
+}
+
+// Test 2: excluding every server leaves selectAnyServer no candidate. It throws a
+// transient, BeforeSend HotRodClientException — and ownersExhausted is FALSE,
+// because a keyless op has no owners (this is the one behavioural difference from
+// the keyed ProxyDisabledOwnersExhaustedThrows case). Deterministic.
+TEST_F(PingRetryIntegrationTest, AllServersExcludedThrowsBeforeSend) {
+    auto cache = createDefaultClient();
+    ASSERT_NE(nullptr, cache);
+
+    auto servers = pingUntilTopologyKnown(cache);
+    ASSERT_GE(servers.size(), 2u);
+
+    bool threw = false;
+    try {
+        cache->excluding(servers).ping().get();
+    } catch (const HotRodClientException& e) {
+        threw = true;
+        EXPECT_EQ(FailurePhase::BeforeSend, e.phase)
+            << "nothing was sent — this is keyless routing exhaustion";
+        EXPECT_FALSE(e.ownersExhausted)
+            << "keyless ops have no owners; ownersExhausted must stay false";
+        EXPECT_TRUE(isTransient(e))
+            << "BeforeSend is always retryable (nothing applied)";
+        for (const auto& s : servers) {
+            EXPECT_NE(std::find(e.triedNodes.begin(), e.triedNodes.end(), s),
+                      e.triedNodes.end())
+                << "excluded server " << s.host << ":" << s.port
+                << " should appear in triedNodes";
+        }
+    }
+    EXPECT_TRUE(threw) << "excluding every server must throw";
+}
+
+// Test 3: kill the server the keyless ping targets first, then drive the
+// documented catch → excluding(e).ping() loop. It recovers on a surviving node.
+// If the first plain ping re-routes automatically (before-send failover) the loop
+// simply succeeds on attempt 0; if it throws, we assert the failure is transient
+// and that the killed node was recorded as tried and then avoided.
+TEST_F(PingRetryIntegrationTest, RecoversAfterFirstCandidateKilled) {
+    auto cache = createDefaultClient();
+    ASSERT_NE(nullptr, cache);
+
+    auto servers = pingUntilTopologyKnown(cache);
+    ASSERT_GE(servers.size(), 2u);
+
+    // selectAnyServer tries the topology in order, so servers.front() is the node
+    // the next keyless ping would hit first.
+    const ServerAddress target = servers.front();
+    const int nodeNum = nodeNumberFor(cache, target);
+    ASSERT_NE(-1, nodeNum) << "first ping candidate must map to a cluster node";
+
+    fprintf(stderr, "[TEST] killing first ping candidate node %d (%s:%u)\n",
+            nodeNum, target.host.c_str(), target.port);
+    removeNode(nodeNum);
+    std::this_thread::sleep_for(std::chrono::seconds(3));
+
+    std::vector<ServerAddress> excluded;
+    bool recovered = false;
+    bool sawException = false;
+
+    for (int attempt = 0; attempt < 4; ++attempt) {
+        try {
+            if (excluded.empty())
+                cache->ping().get();
+            else
+                cache->excluding(excluded).ping().get();
+            recovered = true;
+            break;
+        } catch (const HotRodClientException& e) {
+            sawException = true;
+            ASSERT_TRUE(isTransient(e))
+                << "a killed-node ping failure should be retryable, not futile";
+            ASSERT_GT(e.triedNodes.size(), excluded.size())
+                << "each failed attempt must widen the exclusion set";
+            excluded = e.triedNodes;
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+        }
+    }
+
+    EXPECT_TRUE(recovered) << "the ping retry loop must recover on a surviving node";
+
+    if (sawException) {
+        EXPECT_NE(std::find(excluded.begin(), excluded.end(), target),
+                  excluded.end())
+            << "the killed node should be in the exclusion set we retried with";
+        fprintf(stderr, "[PASS] recovered via excluding(e).ping() retry loop\n");
+    } else {
+        fprintf(stderr, "[PASS] recovered via automatic before-send re-routing\n");
+    }
+
+    // Restore the cluster for subsequent tests.
+    addNode(nodeNum);
+    std::this_thread::sleep_for(std::chrono::seconds(3));
+}
+
+// Main — 3-node cluster, matching the other multi-server suites.
+int main(int argc, char** argv) {
+    ::testing::InitGoogleTest(&argc, argv);
+    MultiServerTestEnvironment::setInitialServerCount(3);
+    ::testing::AddGlobalTestEnvironment(new MultiServerTestEnvironment());
+    return RUN_ALL_TESTS();
+}
diff --git a/tests/integration/RetryViewIntegrationTest.cpp b/tests/integration/RetryViewIntegrationTest.cpp
new file mode 100644
index 0000000..93bf687
--- /dev/null
+++ b/tests/integration/RetryViewIntegrationTest.cpp
@@ -0,0 +1,227 @@
+#include <gtest/gtest.h>
+#include "TopologyTestFixture.h"
+#include "hotrod/RemoteCache.h"
+#include "hotrod/HotRodClientException.h"
+#include <thread>
+#include <chrono>
+#include <optional>
+#include <vector>
+#include <string>
+#include <algorithm>
+
+using namespace hotrod;
+using namespace hotrod::test;
+
+/**
+ * Step 11b — user-decided retry, end-to-end.
+ *
+ * These tests exercise the *RetryView* dispatch path — the bound view returned by
+ * RemoteCache::excluding() — which is how a caller retries an operation while
+ * avoiding the nodes a prior attempt already tried. They are deliberately
+ * distinct from the pre-existing failover tests (ConcurrentMultiServerTest
+ * .ConcurrentWithFailover, HashAwareRoutingIntegrationTest.FailoverToBackupOwner),
+ * which cover the *automatic* before-send re-routing that happens on the next
+ * plain call. Here we assert the explicit, user-driven exclusion API:
+ *
+ *   - excluding({owner}) routes the request around an excluded node (deterministic)
+ *   - proxyToNonOwner=false + all owners excluded throws ownersExhausted (deterministic)
+ *   - the documented catch → excluding(e) retry loop recovers after a node is killed
+ *
+ * See docs/ERROR_HANDLING_DESIGN.md §3.2 / D2 / D5 and docs/DECISIONS.md.
+ */
+class RetryViewIntegrationTest : public TopologyTest {
+protected:
+    void SetUp() override {
+        TopologyTest::SetUp();
+        resetCluster(3);          // hash-aware routing needs a real cluster (numOwners=2)
+        clearTestCache();
+    }
+
+    // Build the pure-value address of a ServerInfo* owner, matching exactly what
+    // orderKeyCandidates()/the exclusion set compare on (host + port).
+    static ServerAddress addrOf(const ServerInfo* s) {
+        return ServerAddress{s->host, s->port};
+    }
+};
+
+// Test 1: excluding({primary owner}) still fetches the value by routing to a
+// different node (a backup owner, or a non-owner proxy). Deterministic — no node
+// is killed; this isolates the RetryView routing decision from failure timing.
+TEST_F(RetryViewIntegrationTest, ExcludingRoutesAroundOwner) {
+    auto cache = createDefaultClient();
+    ASSERT_NE(nullptr, cache);
+
+    std::this_thread::sleep_for(std::chrono::seconds(2));  // let topology arrive
+
+    std::string keyStr = "retryview-around-owner";
+    std::string valueStr = "retryview-around-owner-value";
+    ByteArray key(keyStr.begin(), keyStr.end());
+    ByteArray value(valueStr.begin(), valueStr.end());
+
+    cache->put(key, value).get();
+    std::this_thread::sleep_for(std::chrono::seconds(2));  // replicate to backup
+
+    ASSERT_TRUE(cache->getConsistentHash().hasHashTopology())
+        << "hash-aware routing must be active for this test";
+
+    auto owners = cache->getConsistentHash().getOwners(key, cache->getTopology());
+    ASSERT_GE(owners.size(), 2u) << "DIST_SYNC with numOwners=2 expected";
+
+    const ServerAddress primary = addrOf(owners[0]);
+    fprintf(stderr, "[TEST] excluding primary owner %s:%u\n",
+            primary.host.c_str(), primary.port);
+
+    // Route around the primary explicitly via the RetryView.
+    auto result = cache->excluding({primary}).get(key).get();
+
+    ASSERT_TRUE(result.has_value())
+        << "value must still be reachable via a backup owner / proxy";
+    EXPECT_EQ(value, *result);
+}
+
+// Test 2: with proxy fallback disabled, excluding every owner leaves no candidate
+// and selection throws a transient, BeforeSend HotRodClientException with
+// ownersExhausted=true — deterministic, no node killed. This is the D5 contract:
+// the library refuses to silently proxy and hands the decision back to the caller.
+TEST_F(RetryViewIntegrationTest, ProxyDisabledOwnersExhaustedThrows) {
+    auto cache = createDefaultClient();
+    ASSERT_NE(nullptr, cache);
+
+    std::this_thread::sleep_for(std::chrono::seconds(2));
+
+    std::string keyStr = "retryview-owners-exhausted";
+    std::string valueStr = "retryview-owners-exhausted-value";
+    ByteArray key(keyStr.begin(), keyStr.end());
+    ByteArray value(valueStr.begin(), valueStr.end());
+
+    cache->put(key, value).get();
+    std::this_thread::sleep_for(std::chrono::seconds(2));
+
+    ASSERT_TRUE(cache->getConsistentHash().hasHashTopology());
+
+    auto owners = cache->getConsistentHash().getOwners(key, cache->getTopology());
+    ASSERT_GE(owners.size(), 1u);
+
+    std::vector<ServerAddress> allOwners;
+    for (const auto* o : owners) allOwners.push_back(addrOf(o));
+
+    cache->setProxyToNonOwner(false);  // no non-owner fallback
+
+    bool threw = false;
+    try {
+        cache->excluding(allOwners).get(key).get();
+    } catch (const HotRodClientException& e) {
+        threw = true;
+        EXPECT_EQ(FailurePhase::BeforeSend, e.phase)
+            << "nothing was sent — this is a routing exhaustion";
+        EXPECT_TRUE(e.ownersExhausted)
+            << "every owner was excluded, so owners are exhausted";
+        EXPECT_TRUE(isTransient(e))
+            << "BeforeSend is always retryable (nothing applied)";
+        // triedNodes carries the excluded owners forward for the next attempt.
+        for (const auto& o : allOwners) {
+            EXPECT_NE(std::find(e.triedNodes.begin(), e.triedNodes.end(), o),
+                      e.triedNodes.end())
+                << "excluded owner " << o.host << ":" << o.port
+                << " should appear in triedNodes";
+        }
+    }
+    EXPECT_TRUE(threw) << "excluding all owners with proxy disabled must throw";
+
+    cache->setProxyToNonOwner(true);  // restore default
+}
+
+// Test 3: the documented user retry loop. Kill the primary owner, then drive the
+// catch → excluding(e) loop. It recovers to a live node holding the value. If the
+// first plain GET happens to re-route automatically (before-send failover), the
+// loop simply succeeds on attempt 0; if it throws, we assert the failure is
+// transient and that feeding e.triedNodes back lands the retry on the backup.
+TEST_F(RetryViewIntegrationTest, UserRetryLoopAfterPrimaryKilled) {
+    auto cache = createDefaultClient();
+    ASSERT_NE(nullptr, cache);
+
+    std::this_thread::sleep_for(std::chrono::seconds(2));
+
+    std::string keyStr = "retryview-kill-primary";
+    std::string valueStr = "retryview-kill-primary-value";
+    ByteArray key(keyStr.begin(), keyStr.end());
+    ByteArray value(valueStr.begin(), valueStr.end());
+
+    cache->put(key, value).get();
+    std::this_thread::sleep_for(std::chrono::seconds(2));  // replicate to backup
+
+    ASSERT_TRUE(cache->getConsistentHash().hasHashTopology());
+
+    const ServerInfo* primaryOwner =
+        cache->getConsistentHash().getPrimaryOwner(key, cache->getTopology());
+    ASSERT_NE(nullptr, primaryOwner);
+
+    // Map the primary owner to its node number (topology is ordered node1,2,3…).
+    const auto& topoServers = cache->getTopology().getServers();
+    int primaryNodeNum = -1;
+    for (size_t i = 0; i < topoServers.size(); ++i) {
+        if (topoServers[i].hashId == primaryOwner->hashId) {
+            primaryNodeNum = static_cast<int>(i) + 1;
+            break;
+        }
+    }
+    ASSERT_NE(-1, primaryNodeNum) << "primary owner must be found in topology";
+
+    fprintf(stderr, "[TEST] killing primary owner node %d (%s:%u)\n",
+            primaryNodeNum, primaryOwner->host.c_str(), primaryOwner->port);
+    removeNode(primaryNodeNum);
+    std::this_thread::sleep_for(std::chrono::seconds(3));
+
+    // Canonical user retry loop (ERROR_HANDLING_DESIGN §3.2): retry-free first
+    // call, then opt into exclusion seeded from the caught exception.
+    std::optional<ByteArray> result;
+    std::vector<ServerAddress> excluded;
+    bool sawException = false;
+
+    for (int attempt = 0; attempt < 4; ++attempt) {
+        try {
+            result = excluded.empty()
+                ? cache->get(key).get()
+                : cache->excluding(excluded).get(key).get();
+            break;  // success — either directly or via the exclusion retry
+        } catch (const HotRodClientException& e) {
+            sawException = true;
+            ASSERT_TRUE(isTransient(e))
+                << "a killed-node failure should be retryable, not futile";
+            // GET is idempotent, so replaying is always safe here; a caller with a
+            // non-idempotent op would additionally consult outcomeUncertain(e).
+            ASSERT_GT(e.triedNodes.size(), excluded.size())
+                << "each failed attempt must widen the exclusion set";
+            excluded = e.triedNodes;  // feed tried nodes back as next exclusion
+            std::this_thread::sleep_for(std::chrono::seconds(1));
+        }
+    }
+
+    ASSERT_TRUE(result.has_value())
+        << "the retry loop must recover the value from a surviving owner";
+    EXPECT_EQ(value, *result);
+
+    if (sawException) {
+        // If we went through the catch path, the killed primary must be recorded
+        // as a tried node the retry then avoided.
+        const ServerAddress killed = addrOf(primaryOwner);
+        EXPECT_NE(std::find(excluded.begin(), excluded.end(), killed),
+                  excluded.end())
+            << "the killed primary should be in the exclusion set we retried with";
+        fprintf(stderr, "[PASS] recovered via excluding(e) retry loop\n");
+    } else {
+        fprintf(stderr, "[PASS] recovered via automatic before-send re-routing\n");
+    }
+
+    // Restore the cluster for subsequent tests.
+    addNode(primaryNodeNum);
+    std::this_thread::sleep_for(std::chrono::seconds(3));
+}
+
+// Main — 3-node cluster, matching the other multi-server suites.
+int main(int argc, char** argv) {
+    ::testing::InitGoogleTest(&argc, argv);
+    MultiServerTestEnvironment::setInitialServerCount(3);
+    ::testing::AddGlobalTestEnvironment(new MultiServerTestEnvironment());
+    return RUN_ALL_TESTS();
+}
diff --git a/tests/integration/TopologyTestFixture.h b/tests/integration/TopologyTestFixture.h
index 7bbc094..5240cc0 100644
--- a/tests/integration/TopologyTestFixture.h
+++ b/tests/integration/TopologyTestFixture.h
@@ -9,6 +9,7 @@
 #include <string>
 #include <thread>
 #include <chrono>
+#include <mutex>
 
 namespace hotrod {
 namespace test {
@@ -102,7 +103,16 @@ class TopologyTest : public ::testing::Test {
         client->connect();
 
         RemoteCache* ptr = client.get();
-        clients.push_back(std::move(client));
+        // Tests may call createClient() concurrently from several threads (e.g.
+        // ConcurrentMultiServerTest.MultipleClientsInterlaced), so guard the
+        // shared vector: std::vector::push_back is not safe for concurrent
+        // callers (a reallocation moves the buffer out from under another
+        // thread). This lock keeps TSan runs clean so real production races stay
+        // visible.
+        {
+            std::lock_guard<std::mutex> lock(clientsMutex);
+            clients.push_back(std::move(client));
+        }
 
         return ptr;
     }
@@ -225,8 +235,11 @@ class TopologyTest : public ::testing::Test {
         MultiServerTestEnvironment::waitForClusterSize(currentSize - 1);
     }
 
-    // Storage for created clients (auto-cleanup in TearDown)
+    // Storage for created clients (auto-cleanup in TearDown).
+    // Guarded by clientsMutex because createClient() can be called concurrently;
+    // TearDown runs single-threaded after all worker threads have joined.
     std::vector<std::unique_ptr<RemoteCache>> clients;
+    std::mutex clientsMutex;
 };
 
 } // namespace test
diff --git a/tests/unit/ServerSelectionTest.cpp b/tests/unit/ServerSelectionTest.cpp
new file mode 100644
index 0000000..656deb6
--- /dev/null
+++ b/tests/unit/ServerSelectionTest.cpp
@@ -0,0 +1,195 @@
+#include <gtest/gtest.h>
+#include "hotrod/ServerSelection.h"
+#include <vector>
+
+using namespace hotrod;
+
+/**
+ * Unit tests for orderKeyCandidates() — the pure routing decision behind
+ * exclusion-aware selectServerForKey (Step 11b).
+ *
+ * It has no I/O, so the ordering/exclusion/dedup behaviour is tested directly,
+ * independent of connection acquisition.
+ *
+ * See docs/ERROR_HANDLING_DESIGN.md §3.2.
+ */
+
+namespace {
+ServerAddress addr(const std::string& host, uint16_t port) { return {host, port}; }
+
+std::vector<std::string> asKeys(const std::vector<ServerAddress>& v) {
+    std::vector<std::string> out;
+    for (const auto& a : v) out.push_back(a.host + ":" + std::to_string(a.port));
+    return out;
+}
+} // namespace
+
+// Owners come first, in order, then non-owner servers as proxy fallback.
+TEST(ServerSelectionTest, OwnersFirstThenNonOwnerFallback) {
+    std::vector<ServerAddress> owners = {addr("a", 1), addr("b", 2)};
+    std::vector<ServerAddress> all = {addr("a", 1), addr("b", 2), addr("c", 3), addr("d", 4)};
+
+    auto out = orderKeyCandidates(owners, all, /*exclude=*/{});
+
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"a:1", "b:2", "c:3", "d:4"}));
+}
+
+// A server that is both an owner and in allServers appears once, in owner order.
+TEST(ServerSelectionTest, DeduplicatesOwnersAgainstAllServers) {
+    std::vector<ServerAddress> owners = {addr("b", 2)};
+    std::vector<ServerAddress> all = {addr("a", 1), addr("b", 2), addr("c", 3)};
+
+    auto out = orderKeyCandidates(owners, all, {});
+
+    // b:2 kept at its owner position (front), not duplicated when seen in `all`.
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"b:2", "a:1", "c:3"}));
+}
+
+// Excluded nodes are dropped whether they are owners or non-owners.
+TEST(ServerSelectionTest, ExcludesTriedNodes) {
+    std::vector<ServerAddress> owners = {addr("a", 1), addr("b", 2)};
+    std::vector<ServerAddress> all = {addr("a", 1), addr("b", 2), addr("c", 3)};
+
+    auto out = orderKeyCandidates(owners, all, /*exclude=*/{addr("a", 1), addr("c", 3)});
+
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"b:2"}));
+}
+
+// Excluding every candidate yields an empty list (caller treats as exhausted).
+TEST(ServerSelectionTest, AllExcludedYieldsEmpty) {
+    std::vector<ServerAddress> owners = {addr("a", 1)};
+    std::vector<ServerAddress> all = {addr("a", 1), addr("b", 2)};
+
+    auto out = orderKeyCandidates(owners, all, {addr("a", 1), addr("b", 2)});
+
+    EXPECT_TRUE(out.empty());
+}
+
+// Port distinguishes servers on the same host.
+TEST(ServerSelectionTest, SameHostDifferentPortAreDistinct) {
+    std::vector<ServerAddress> owners = {addr("h", 11222)};
+    std::vector<ServerAddress> all = {addr("h", 11222), addr("h", 11322)};
+
+    auto out = orderKeyCandidates(owners, all, {addr("h", 11222)});
+
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"h:11322"}));
+}
+
+// A non-owner fallback still appears when all owners are excluded.
+TEST(ServerSelectionTest, NonOwnerFallbackWhenOwnersExcluded) {
+    std::vector<ServerAddress> owners = {addr("a", 1)};
+    std::vector<ServerAddress> all = {addr("a", 1), addr("b", 2), addr("c", 3)};
+
+    auto out = orderKeyCandidates(owners, all, {addr("a", 1)});
+
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"b:2", "c:3"}));
+}
+
+// Empty topology yields no candidates even with owners listed.
+TEST(ServerSelectionTest, EmptyAllServersKeepsOwnersOnly) {
+    std::vector<ServerAddress> owners = {addr("a", 1), addr("b", 2)};
+
+    auto out = orderKeyCandidates(owners, /*all=*/{}, {});
+
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"a:1", "b:2"}));
+}
+
+// proxyToNonOwner=true (default) appends non-owner servers after the owners.
+TEST(ServerSelectionTest, ProxyToNonOwnerTrueAppendsNonOwners) {
+    std::vector<ServerAddress> owners = {addr("a", 1)};
+    std::vector<ServerAddress> all = {addr("a", 1), addr("b", 2), addr("c", 3)};
+
+    auto out = orderKeyCandidates(owners, all, /*exclude=*/{}, /*proxyToNonOwner=*/true);
+
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"a:1", "b:2", "c:3"}));
+}
+
+// proxyToNonOwner=false stops at the owners — no non-owner fallback.
+TEST(ServerSelectionTest, ProxyToNonOwnerFalseKeepsOwnersOnly) {
+    std::vector<ServerAddress> owners = {addr("a", 1)};
+    std::vector<ServerAddress> all = {addr("a", 1), addr("b", 2), addr("c", 3)};
+
+    auto out = orderKeyCandidates(owners, all, /*exclude=*/{}, /*proxyToNonOwner=*/false);
+
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"a:1"}));
+}
+
+// proxyToNonOwner=false with every owner excluded yields no candidates — the
+// caller then reports ownersExhausted rather than proxying.
+TEST(ServerSelectionTest, ProxyToNonOwnerFalseAllOwnersExcludedYieldsEmpty) {
+    std::vector<ServerAddress> owners = {addr("a", 1), addr("b", 2)};
+    std::vector<ServerAddress> all = {addr("a", 1), addr("b", 2), addr("c", 3)};
+
+    auto out = orderKeyCandidates(owners, all, /*exclude=*/{addr("a", 1), addr("b", 2)},
+                                  /*proxyToNonOwner=*/false);
+
+    EXPECT_TRUE(out.empty());
+}
+
+// --- keyless candidate ordering (Step 11c): the empty-owner list that
+//     selectAnyServer() relies on for ping and other non-key operations. ---
+
+// With no owners (keyless op), every server is a candidate in topology order.
+TEST(ServerSelectionTest, KeylessNoOwnersYieldsAllServersInOrder) {
+    std::vector<ServerAddress> all = {addr("a", 1), addr("b", 2), addr("c", 3)};
+
+    auto out = orderKeyCandidates(/*owners=*/{}, all, /*exclude=*/{});
+
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"a:1", "b:2", "c:3"}));
+}
+
+// Keyless selection still honors the exclusion set, so a retry avoids the node a
+// prior attempt tried.
+TEST(ServerSelectionTest, KeylessExcludesTriedNodes) {
+    std::vector<ServerAddress> all = {addr("a", 1), addr("b", 2), addr("c", 3)};
+
+    auto out = orderKeyCandidates(/*owners=*/{}, all, /*exclude=*/{addr("b", 2)});
+
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"a:1", "c:3"}));
+}
+
+// Keyless with every server excluded yields no candidates — selectAnyServer then
+// throws a BeforeSend exhaustion (ownersExhausted=false for a keyless op).
+TEST(ServerSelectionTest, KeylessAllExcludedYieldsEmpty) {
+    std::vector<ServerAddress> all = {addr("a", 1), addr("b", 2)};
+
+    auto out = orderKeyCandidates(/*owners=*/{}, all,
+                                  /*exclude=*/{addr("a", 1), addr("b", 2)});
+
+    EXPECT_TRUE(out.empty());
+}
+
+// --- unionNodes: triedNodes accumulation across retries (Step 11b) ---
+
+// Union appends only new members, preserving the first list's order.
+TEST(ServerSelectionTest, UnionAppendsNewNodesPreservingOrder) {
+    auto out = unionNodes({addr("a", 1), addr("b", 2)}, {addr("c", 3)});
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"a:1", "b:2", "c:3"}));
+}
+
+// Members already present are not duplicated.
+TEST(ServerSelectionTest, UnionDeduplicates) {
+    auto out = unionNodes({addr("a", 1), addr("b", 2)}, {addr("b", 2), addr("c", 3)});
+    EXPECT_EQ(asKeys(out), (std::vector<std::string>{"a:1", "b:2", "c:3"}));
+}
+
+// Empty operands are handled both ways.
+TEST(ServerSelectionTest, UnionWithEmptyOperands) {
+    EXPECT_EQ(asKeys(unionNodes({}, {addr("a", 1)})), (std::vector<std::string>{"a:1"}));
+    EXPECT_EQ(asKeys(unionNodes({addr("a", 1)}, {})), (std::vector<std::string>{"a:1"}));
+    EXPECT_TRUE(unionNodes({}, {}).empty());
+}
+
+// Folding successive attempts' tried-sets grows monotonically without dupes —
+// exactly the accumulation a user's retry loop performs (excludeNodes grows).
+TEST(ServerSelectionTest, UnionAccumulatesAcrossRetries) {
+    std::vector<ServerAddress> excluded;                       // attempt 1: nothing excluded
+    excluded = unionNodes(excluded, {addr("a", 1)});           // attempt 1 tried a
+    EXPECT_EQ(asKeys(excluded), (std::vector<std::string>{"a:1"}));
+
+    excluded = unionNodes(excluded, {addr("a", 1), addr("b", 2)}); // attempt 2 tried a(skip)+b
+    EXPECT_EQ(asKeys(excluded), (std::vector<std::string>{"a:1", "b:2"}));
+
+    excluded = unionNodes(excluded, {addr("c", 3)});          // attempt 3 tried c
+    EXPECT_EQ(asKeys(excluded), (std::vector<std::string>{"a:1", "b:2", "c:3"}));
+}