From 601216e35d8b8e014454b2ad827883c36f327efe Mon Sep 17 00:00:00 2001
From: Igor Melnichenko <i-melnichenko@yandex-team.ru>
Date: Tue, 4 Aug 2026 23:23:45 +0300
Subject: [PATCH] Read endpoint pool state under the lock in pessimizeEndpoint

pessimizeEndpoint looked up recordsByEndpoint and tested
knownEndpoint.isPessimized() before taking any lock. Both the map
reference and the priority field it reads are non-volatile and are
replaced or written under the write lock by setNewState and pessimize, so
those reads had no happens-before edge with the writers. Two threads
failing on the same endpoint could each miss the other's update and
redo the full re-sort and rescan of the pool, and a lookup against a
stale map could pessimize an entry that is no longer part of the current
state.

Keep the cheap "already pessimized" short circuit but perform it under the
read lock, then re-validate under the write lock before mutating.

needToRunDiscovery is written under the write lock but read by the
discovery scheduler thread with no lock at all, so a pessimization-driven
discovery could go unnoticed indefinitely. Make it volatile.

This is a memory visibility fix; the interleavings involved are not
reproducible deterministically, so it comes without a new test.
---
 .../tech/ydb/core/impl/pool/EndpointPool.java | 31 +++++++++++++------
 1 file changed, 22 insertions(+), 9 deletions(-)

diff --git a/core/src/main/java/tech/ydb/core/impl/pool/EndpointPool.java b/core/src/main/java/tech/ydb/core/impl/pool/EndpointPool.java
index 7cd15e878..9a1f07803 100644
--- a/core/src/main/java/tech/ydb/core/impl/pool/EndpointPool.java
+++ b/core/src/main/java/tech/ydb/core/impl/pool/EndpointPool.java
@@ -42,7 +42,8 @@ public final class EndpointPool {
     private Map<Integer, PriorityEndpoint> recordsByNodeId = new HashMap<>();
     private Map<String, PriorityEndpoint> recordsByEndpoint = new HashMap<>();
 
-    private boolean needToRunDiscovery = false;
+    // read by the discovery scheduler thread without the lock
+    private volatile boolean needToRunDiscovery = false;
     // Number of endpoints with best load factor (priority)
     private int bestEndpointsCount = -1;
 
@@ -163,18 +164,30 @@ public void pessimizeEndpoint(EndpointRecord endpoint, String reason) {
             return;
         }
 
-        PriorityEndpoint knownEndpoint = recordsByEndpoint.get(endpoint.getHostAndPort());
-        if (knownEndpoint == null) {
-            return;
-        }
-
-        if (knownEndpoint.isPessimized()) {
-            logger.trace("Endpoint {} is already pessimized", endpoint);
-            return;
+        // Fast path under the read lock: re-pessimizing an endpoint changes nothing but forces a full
+        // re-sort of the pool
+        recordsLock.readLock().lock();
+        try {
+            PriorityEndpoint knownEndpoint = recordsByEndpoint.get(endpoint.getHostAndPort());
+            if (knownEndpoint == null) {
+                return;
+            }
+            if (knownEndpoint.isPessimized()) {
+                logger.trace("Endpoint {} is already pessimized", endpoint);
+                return;
+            }
+        } finally {
+            recordsLock.readLock().unlock();
         }
 
         recordsLock.writeLock().lock();
         try {
+            // the pool state may have been replaced by setNewState between the two locks
+            PriorityEndpoint knownEndpoint = recordsByEndpoint.get(endpoint.getHostAndPort());
+            if (knownEndpoint == null || knownEndpoint.isPessimized() || records.isEmpty()) {
+                return;
+            }
+
             knownEndpoint.pessimize();
 
             records.sort(PriorityEndpoint.COMPARATOR);