diff --git a/CHANGELOG.md b/CHANGELOG.md index de4ff77c..1d9ee0b0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,31 @@ terminates with an error. Providing not positive timeout value throws an exception. - Fixed connection close callbacks executing before the connection state becomes closed. - Fixed greeting object remaining non-null after connection closure. +- Fixed a bug when a server that accepts the TCP connection but does not send the greeting was + reported as "Connection closed by server". The connect future is now failed only with + `TimeoutException`, whose message names the address and the timeout, and no + `CLOSE_BY_REMOTE` event is fired. Previously the connection pool handled every such attempt + twice, so the reconnect chains multiplied while the server stayed silent. The greeting timer of + an attempt that the server closed no longer closes the channel of the next attempt. +- Fixed reconnect chains multiplying when a server (or a proxy in front of it) accepts the TCP + connection and closes it before the greeting: the close was reported both by the failed connect + future and by a `CLOSE_BY_REMOTE` event, and the pool handled each report. Close events are now + fired only for a connection that has received the greeting. +- Fixed a leak when graceful shutdown is enabled: every connect attempt added one more + `box.shutdown` callback to the same watcher, so a client that kept reconnecting accumulated + callbacks, and a `box.shutdown` event after a successful connect ran the callback once per + previous attempt. The callback is now added once per client. +- Fixed the `pool.connect.time` timer of the connection pool: it was stopped only when an + attempt succeeded, so every failed attempt left an active sample behind. +- Fixed the count of unavailable connections in the pool drifting when a connection was locked + or unlocked from several threads at once (connect, close and heartbeat handlers): both calls + could pass the check and change the count twice. A count below the real value made the + balancer loop while all connections were locked; a count above it hid free connections or made + the pool fail with "Invalid pool state". +- Fixed `reconnectAfter` being ignored for the connections created with the client: the client + builder set it on the pool after the connections had been created with the default of 1000 ms, + and the pool did not pass the new value to existing connections. `withReconnectAfter` and the + `reconnect-after` property of Spring Data now take effect. ### Documentation @@ -28,6 +53,13 @@ - Added the "Limitations and workarounds" page: no batch delete and no `IN` condition in `crud`, examples of parallel requests instead, and derived-method keywords that are not supported. - Added a table that maps Spring Boot minor lines to `tarantool-spring-data-*` modules. +- Described locked connections of the pool: when a connection is locked and unlocked, and that a + request fails with `NoAvailableClientsException` when all connections are locked. + +### Testcontainers + +- Fixed the description of the `testcontainers` module in its `pom.xml`: it was copied from the + connector. ## [1.7.2] - 2026-09-25 diff --git a/documentation/doc-src/pages/client/arch/connection-pool.en.md b/documentation/doc-src/pages/client/arch/connection-pool.en.md index 9f5d59a3..162150c0 100644 --- a/documentation/doc-src/pages/client/arch/connection-pool.en.md +++ b/documentation/doc-src/pages/client/arch/connection-pool.en.md @@ -28,3 +28,16 @@ requests can be found in the [balancers](balancer.md) section. The actual establishment of the connection in `IProtoClientPool` occurs when the balancer first selects this connection from the pool. + +### Locked connections + +A `PoolEntry` is *locked* when its connection is considered unusable: a connection attempt failed, +the established connection was closed by the server or by a server shutdown, or the +[heartbeat](heartbeat.md) moved the connection to the invalidated state. A locked entry is not +returned to the balancer and is unlocked when the connection is restored (the reconnect succeeds or +the heartbeat reports recovery) or when the entry is removed from a shrinking group. A connection is **not** locked just because a long request is +running on it: many requests are in flight over one iproto connection at the same time. + +The pool counts locked entries. When all entries are locked, the balancer does not wait for one to +be unlocked: the request fails with `NoAvailableClientsException`. A server that is busy so long +that it stops answering pings can lead to this, because the heartbeat invalidates its connections. diff --git a/documentation/doc-src/pages/client/arch/connection-pool.md b/documentation/doc-src/pages/client/arch/connection-pool.md index 40535ec4..0abebc25 100644 --- a/documentation/doc-src/pages/client/arch/connection-pool.md +++ b/documentation/doc-src/pages/client/arch/connection-pool.md @@ -30,3 +30,17 @@ title: Пул соединений Фактическое установление соединения в `IProtoClientPool` происходит, когда балансировщик впервые выбирает это соединение из пула. + +### Заблокированные соединения + +`PoolEntry` *заблокирован*, когда его соединение считается непригодным: попытка подключения +завершилась ошибкой, установленное соединение закрыл сервер или оно закрылось при выключении +сервера, либо [heartbeat](heartbeat.md) перевёл соединение в состояние invalidated. Заблокированная +запись не отдаётся балансировщику и разблокируется, когда соединение восстановлено (переподключение +удалось или heartbeat сообщил о восстановлении) или когда запись удаляется из уменьшаемой группы. Соединение **не** блокируется из-за того, что на нём +выполняется долгий запрос: по одному соединению iproto одновременно идёт много запросов. + +Пул ведёт счёт заблокированных записей. Если заблокированы все записи, балансировщик не ждёт, пока +какая-нибудь освободится: запрос завершается с `NoAvailableClientsException`. К этому может привести +сервер, настолько долго занятый, что не отвечает на пинги: heartbeat признаёт его соединения +недействительными. diff --git a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java index b248c732..fa8c213c 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java +++ b/tarantool-core/src/main/java/io/tarantool/core/IProtoClientImpl.java @@ -16,6 +16,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -97,6 +98,10 @@ public class IProtoClientImpl implements IProtoClient { private final List toUnwatch; protected final Map fsmRegistry; private final Map watchers; + + /** Set once the {@code box.shutdown} callback is added; it survives reconnects. */ + private final AtomicBoolean shutdownWatcherRegistered = new AtomicBoolean(); + protected final Timer timerService; private final WatcherOptions watcherOpts; private CompletableFuture serverProtocolVersion; @@ -185,7 +190,7 @@ public CompletableFuture connect(InetSocketAddress address, long timeoutMs @Override public CompletableFuture connect( InetSocketAddress address, long timeoutMs, boolean gracefulShutdown) { - if (gracefulShutdown) { + if (gracefulShutdown && shutdownWatcherRegistered.compareAndSet(false, true)) { watch(SHUTDOWN_EVENT_KEY, this::shutdownEventCallback); } diff --git a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java index b3d31e40..15f31ba0 100644 --- a/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java +++ b/tarantool-core/src/main/java/io/tarantool/core/connection/ConnectionImpl.java @@ -44,7 +44,9 @@ private enum CloseReason { /** Closed by client close reason. */ CLOSED_BY_CLIENT(ERR_CLOSED_BY_CLIENT), /** Closed by shutdown close reason. */ - CLOSED_BY_SHUTDOWN(ERR_CLOSED_BY_SHUTDOWN); + CLOSED_BY_SHUTDOWN(ERR_CLOSED_BY_SHUTDOWN), + /** Closed by the client because the greeting did not arrive in time. */ + CLOSED_BY_TIMEOUT(ERR_CLOSED_BY_TIMEOUT); private final String message; @@ -98,6 +100,9 @@ public String getMessage() { private static final String ERR_CLOSED_BY_CLIENT = "Connection closed by client"; private static final String ERR_CLOSED_BY_SERVER = "Connection closed by server"; private static final String ERR_CLOSED_BY_SHUTDOWN = "Connection closed by shutdown"; + private static final String ERR_CLOSED_BY_TIMEOUT = "Connection closed by connect timeout"; + private static final String ERR_GREETING_TIMEOUT = + "Connection timeout: no greeting from %s within %d ms"; private static final String ERR_NOT_CONNECTED = "Connection is not established"; private static final String ERR_SEND_FAILURE = "Failed to send IProto message: %s"; @@ -118,6 +123,16 @@ public String getMessage() { private Consumer consumer; private CompletableFuture connectPromise; private Timeout timeoutHandler; + + /** Greeting promise of the current attempt; a timer of an earlier attempt must not touch it. */ + private CompletableFuture greetingPromise; + + /** + * Whether the current attempt has received the greeting. A close of a connection that has never + * been established is reported only through the failed connect future, not by a close event. + */ + private volatile boolean established; + private int idleTimeout; /** @@ -190,13 +205,16 @@ public synchronized CompletableFuture connect(InetSocketAddress addres log.debug("{} is connecting to {}", this, address); CompletableFuture promise = new CompletableFuture<>(); - scheduleGreetingTimeout(promise, timeoutMs); + greetingPromise = promise; + established = false; + scheduleGreetingTimeout(promise, address, timeoutMs); connectPromise = promise.thenApply( greeting -> { cancelGreetingTimeout(); log.info("{} connected to {}", this, address); this.greeting = greeting; + established = true; state.set(State.READY); return greeting; }); @@ -358,6 +376,15 @@ private synchronized void onChannelClose(ChannelFuture f) { String message; ConnectionCloseEvent event; CloseReason lastCloseReason = closedBy.get(); + if (lastCloseReason == CloseReason.CLOSED_BY_TIMEOUT) { + // The greeting timer has already failed the connect promise with TimeoutException, so the + // greeting can no longer complete it: the connection has never been established and there + // is no close event to fire. + state.set(State.CLOSED); + channel = null; + connectPromise = null; + return; + } switch (lastCloseReason) { case CLOSED_BY_SHUTDOWN: message = lastCloseReason.getMessage(); @@ -377,13 +404,14 @@ private synchronized void onChannelClose(ChannelFuture f) { if (notify) { failConnectPromise(ex); } + final boolean fireEvent = notify && established; channel = null; connectPromise = null; greeting = null; state.set(State.CLOSED); - if (notify) { + if (fireEvent) { fire(event, ex); } } @@ -416,23 +444,42 @@ private void fire(ConnectionCloseEvent event, Throwable ex) { } } - private void scheduleGreetingTimeout(CompletableFuture promise, long timeoutMs) { + private void scheduleGreetingTimeout( + CompletableFuture promise, InetSocketAddress address, long timeoutMs) { log.debug("Schedule timeout for connect in milliseconds: {}", timeoutMs); timeoutHandler = this.timerService.newTimeout( timeoutHandler -> { - if (promise.isDone()) { - return; - } - if (state.get() == State.CONNECTING || state.get() == State.WAITING) { - channel.pipeline().close(); - promise.completeExceptionally(new TimeoutException("Connection timeout")); + // Fail the promise before closing the channel: if the greeting has already + // completed it, the connection is established and must not be closed here. + if (promise.completeExceptionally( + new TimeoutException(String.format(ERR_GREETING_TIMEOUT, address, timeoutMs)))) { + closeOnGreetingTimeout(promise); } }, timeoutMs, TimeUnit.MILLISECONDS); } + /** + * Closes the channel of a connection whose greeting promise has been failed by the timeout. + * + *

The close reason is set before the channel is closed, so the close is not reported as a + * close by server: the connection has never been established. A timer of an earlier attempt does + * nothing: its channel is already closed, and the current one belongs to another attempt. + * + * @param promise greeting promise of the attempt the timer was scheduled for + */ + private synchronized void closeOnGreetingTimeout(CompletableFuture promise) { + if (promise != greetingPromise) { + return; + } + if (state.get() != State.CONNECTING && state.get() != State.WAITING) { + return; + } + closeChannel(CloseReason.CLOSED_BY_TIMEOUT); + } + private void cancelGreetingTimeout() { if (timeoutHandler == null) { return; diff --git a/tarantool-core/src/test/java/io/tarantool/core/ShutdownWatcherRegistrationTest.java b/tarantool-core/src/test/java/io/tarantool/core/ShutdownWatcherRegistrationTest.java new file mode 100644 index 00000000..d4d68d7f --- /dev/null +++ b/tarantool-core/src/test/java/io/tarantool/core/ShutdownWatcherRegistrationTest.java @@ -0,0 +1,90 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core; + +import java.lang.reflect.Field; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.tarantool.core.connection.ConnectionFactory; + +/** Reconnect attempts must not add a new {@code box.shutdown} callback each time. */ +public class ShutdownWatcherRegistrationTest { + + private static final int ATTEMPTS = 5; + + private EventLoopGroup group; + private Timer timer; + private IProtoClientImpl client; + + @BeforeEach + public void setUp() { + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + timer = new HashedWheelTimer(); + Bootstrap bootstrap = + new Bootstrap() + .group(group) + .channel(NioSocketChannel.class) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000); + client = new IProtoClientImpl(new ConnectionFactory(bootstrap, timer), timer); + } + + @AfterEach + public void tearDown() throws Exception { + client.close(); + timer.stop(); + group.shutdownGracefully().get(); + } + + @Test + public void test_failedConnects_shouldRegisterShutdownCallbackOnce() throws Exception { + InetSocketAddress refused = closedPortAddress(); + + for (int i = 0; i < ATTEMPTS; i++) { + CompletableFuture attempt = client.connect(refused, 1000, true); + assertThrows(ExecutionException.class, () -> attempt.get(5, TimeUnit.SECONDS)); + } + + assertEquals(1, shutdownCallbacksCount()); + } + + private static InetSocketAddress closedPortAddress() throws Exception { + try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + return new InetSocketAddress(socket.getInetAddress(), socket.getLocalPort()); + } + } + + // The callbacks are not exposed by the API, so the test reads them directly. + @SuppressWarnings("unchecked") + private int shutdownCallbacksCount() throws Exception { + Field watchersField = IProtoClientImpl.class.getDeclaredField("watchers"); + watchersField.setAccessible(true); + Map watchers = (Map) watchersField.get(client); + Field callbacksField = Watcher.class.getDeclaredField("callbacks"); + callbacksField.setAccessible(true); + return ((List) callbacksField.get(watchers.get("box.shutdown"))).size(); + } +} diff --git a/tarantool-core/src/test/java/io/tarantool/core/connection/GreetingTimeoutTest.java b/tarantool-core/src/test/java/io/tarantool/core/connection/GreetingTimeoutTest.java new file mode 100644 index 00000000..5ea3e64f --- /dev/null +++ b/tarantool-core/src/test/java/io/tarantool/core/connection/GreetingTimeoutTest.java @@ -0,0 +1,158 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core.connection; + +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** A server accepts the TCP connection but never sends the greeting. */ +public class GreetingTimeoutTest { + + private static final long CONNECT_TIMEOUT_MS = 300; + private static final long EVENT_WAIT_MS = 1000; + + private EventLoopGroup group; + private Timer timer; + private ServerSocket silentServer; + private ConnectionImpl connection; + + @BeforeEach + public void setUp() throws Exception { + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + timer = new HashedWheelTimer(); + // The kernel completes the TCP handshake from the backlog; nobody reads or writes. + silentServer = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + Bootstrap bootstrap = + new Bootstrap() + .group(group) + .channel(NioSocketChannel.class) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000); + connection = new ConnectionImpl(bootstrap, timer); + } + + @AfterEach + public void tearDown() throws Exception { + connection.close(); + silentServer.close(); + timer.stop(); + group.shutdownGracefully().get(); + } + + @Test + public void test_greetingTimeout_shouldNotBeReportedAsCloseByServer() throws Exception { + List closedByRemote = new CopyOnWriteArrayList<>(); + connection.onClose(ConnectionCloseEvent.CLOSE_BY_REMOTE, (c, ex) -> closedByRemote.add(ex)); + + InetSocketAddress address = + new InetSocketAddress(silentServer.getInetAddress(), silentServer.getLocalPort()); + CompletableFuture connect = connection.connect(address, CONNECT_TIMEOUT_MS); + + ExecutionException ex = + assertThrows(ExecutionException.class, () -> connect.get(5, TimeUnit.SECONDS)); + assertInstanceOf(TimeoutException.class, ex.getCause()); + assertEquals( + String.format( + "Connection timeout: no greeting from %s within %d ms", address, CONNECT_TIMEOUT_MS), + ex.getCause().getMessage()); + + // Without the fix the close event follows the timeout within milliseconds. + long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(EVENT_WAIT_MS); + while (System.nanoTime() < deadline && closedByRemote.isEmpty()) { + Thread.sleep(10); + } + assertEquals( + 0, + closedByRemote.size(), + "greeting timeout is reported as closed by server: " + closedByRemote); + assertFalse(connection.isConnected()); + } + + @Test + public void test_greetingTimerOfClosedAttempt_shouldNotCloseNextAttempt() throws Exception { + long timeoutMs = 600; + long pauseMs = 300; + try (ServerSocket closingServer = new ServerSocket(0, 50, InetAddress.getLoopbackAddress())) { + // Accepts and closes at once, before the greeting. + Thread closer = + new Thread( + () -> { + try (Socket socket = closingServer.accept()) { + // closed by try-with-resources + } catch (Exception ignored) { + // the test fails on its own assertions + } + }); + closer.setDaemon(true); + closer.start(); + + CompletableFuture first = + connection.connect( + new InetSocketAddress(closingServer.getInetAddress(), closingServer.getLocalPort()), + timeoutMs); + assertThrows(ExecutionException.class, () -> first.get(5, TimeUnit.SECONDS)); + } + Thread.sleep(pauseMs); + + // The silent server records when the client closed its side of the second attempt. + AtomicLong closedAfterMs = new AtomicLong(-1); + long start = System.nanoTime(); + Thread reader = + new Thread( + () -> { + try (Socket socket = silentServer.accept()) { + while (socket.getInputStream().read() != -1) { + // the client never writes + } + } catch (Exception ignored) { + // a reset also ends the connection + } + closedAfterMs.set(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)); + }); + reader.setDaemon(true); + reader.start(); + + CompletableFuture second = + connection.connect( + new InetSocketAddress(silentServer.getInetAddress(), silentServer.getLocalPort()), + timeoutMs); + ExecutionException ex = + assertThrows(ExecutionException.class, () -> second.get(5, TimeUnit.SECONDS)); + assertInstanceOf(TimeoutException.class, ex.getCause()); + reader.join(5000); + + // Without the fix the timer of the first attempt closes the second one about timeoutMs - + // pauseMs after it starts, without any error for it. + assertTrue( + closedAfterMs.get() >= timeoutMs - 100, + "second attempt was closed after " + closedAfterMs.get() + " ms"); + } +} diff --git a/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java b/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java index 77ea1ee4..187d3ea7 100644 --- a/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java +++ b/tarantool-core/src/test/java/io/tarantool/core/integration/ConnectionToTarantoolTest.java @@ -249,7 +249,7 @@ public void testConnectToAddressWithBadHost() { cause.getMessage()); assertEquals(ConnectTimeoutException.class, findRootCause(ex).getClass()); } else if (cause.getClass() == TimeoutException.class) { - assertEquals("Connection timeout", cause.getMessage()); + assertTrue(cause.getMessage().startsWith("Connection timeout"), cause.getMessage()); log.info(findRootCause(ex).getClass().toString()); // Need for debug in the future } else { fail("An unknown exception was thrown during connection to the bad hostname!", cause); @@ -322,7 +322,7 @@ public void testConnectToSilentNode() { Exception ex = assertThrows(CompletionException.class, future::join); Throwable cause = findRootCause(ex); assertEquals(TimeoutException.class, cause.getClass()); - assertEquals("Connection timeout", cause.getMessage()); + assertTrue(cause.getMessage().startsWith("Connection timeout"), cause.getMessage()); } /* @@ -408,7 +408,7 @@ public void testConnectWithGreetingTimeout() throws InterruptedException { assertTrue( cause instanceof TimeoutException || cause instanceof ConnectionClosedException, "Exception should be either TimeoutException or ConnectionClosedException"); - assertEquals("Connection timeout", cause.getMessage()); + assertTrue(cause.getMessage().startsWith("Connection timeout"), cause.getMessage()); } @Test diff --git a/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java b/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java index a4f85cc5..f60880fe 100644 --- a/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java +++ b/tarantool-pooling/src/main/java/io/tarantool/pool/IProtoClientPoolImpl.java @@ -501,6 +501,8 @@ public void setReconnectAfter(long reconnectAfter) throws IllegalArgumentExcepti if (reconnectAfter <= 0) { throw new IllegalArgumentException("reconnect period should be positive number"); } + entries.forEach( + (tag, entryGroup) -> entryGroup.forEach(pe -> pe.setReconnectAfter(reconnectAfter))); this.reconnectAfter = reconnectAfter; } diff --git a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java index ef8b34ad..a70e31e5 100644 --- a/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java +++ b/tarantool-pooling/src/main/java/io/tarantool/pool/PoolEntry.java @@ -8,6 +8,7 @@ import java.util.ArrayDeque; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; import io.micrometer.core.instrument.Counter; @@ -97,7 +98,7 @@ final class PoolEntry { private final int index; /** Time in milliseconds after that reconnect task will be executed. */ - private final long reconnectAfter; + private volatile long reconnectAfter; /** Parameter that will say tarantool not to use Tuple Extension. */ private final boolean useTupleExtension; @@ -132,9 +133,11 @@ final class PoolEntry { * Flag signaling if connection is available or not. * *

When connection comes to invalidated state or killed, pool entry is locked and connection - * will not be returned to outer client. + * will not be returned to outer client. Connect, close and heartbeat handlers run on different + * threads, so the flag is changed with compare-and-set: only the call that changes it updates the + * count of unavailable connections. */ - private boolean isLocked; + private final AtomicBoolean isLocked = new AtomicBoolean(); /** Count of failed pings occurred in invalidated state. */ private int currentDeathPings; @@ -221,7 +224,6 @@ public PoolEntry( this.gracefulShutdown = gracefulShutdown; this.group = group; this.index = index; - this.isLocked = false; this.reconnectAfter = reconnectAfter; this.tag = group.getTag(); this.timerService = timerService; @@ -251,9 +253,8 @@ public IProtoClient getClient() { *

Also increments count of unavailable clients. */ public void lock() { - if (!isLocked) { + if (isLocked.compareAndSet(false, true)) { this.connectionStatistics.incrementUnavailable(); - isLocked = true; } } @@ -263,10 +264,9 @@ public void lock() { *

Also decrements count of unavailable clients and cancels reconnect task. */ public void unlock() { - if (isLocked) { + if (isLocked.compareAndSet(true, false)) { stopReconnectTask(); this.connectionStatistics.decrementUnavailable(); - isLocked = false; } } @@ -276,7 +276,7 @@ public void unlock() { * @return {@link #isLocked} value. */ public boolean isLocked() { - return isLocked; + return isLocked.get(); } /** Closes client and stops heartbeat and reconnect tasks if started. */ @@ -318,6 +318,15 @@ public void setConnectTimeout(long timeout) { this.connectTimeout = timeout; } + /** + * Setter for {@link #reconnectAfter}. + * + * @param reconnectAfter reconnect time in milliseconds + */ + public void setReconnectAfter(long reconnectAfter) { + this.reconnectAfter = reconnectAfter; + } + /** Heartbeat starter. */ public void startHeartbeat() { if (isHeartbeatStarted || heartbeatOpts == null) { @@ -363,12 +372,13 @@ private CompletableFuture internalConnect() { this.metrics != null ? this.metrics.poolConnectTime.start() : null; CompletableFuture future = client.connect(group.getAddress(), connectTimeout, gracefulShutdown); + // Stop the timer on failure too, otherwise every failed attempt leaves an active sample. + future.whenComplete((r, exc) -> PoolEntryMetrics.stopTimer(timer)); String user = group.getUser(); connectFuture = future .thenCompose( greeting -> { - PoolEntryMetrics.stopTimer(timer); if (user != null) { return client.authorize(user, group.getPassword(), group.getAuthType()); } diff --git a/tarantool-pooling/src/test/java/io/tarantool/pool/PoolEntryLockTest.java b/tarantool-pooling/src/test/java/io/tarantool/pool/PoolEntryLockTest.java new file mode 100644 index 00000000..5eac092b --- /dev/null +++ b/tarantool-pooling/src/test/java/io/tarantool/pool/PoolEntryLockTest.java @@ -0,0 +1,113 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.pool; + +import java.net.InetAddress; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.tarantool.core.connection.ConnectionFactory; + +/** Lock and unlock of one pool entry called from several threads at once. */ +public class PoolEntryLockTest { + + private static final int THREADS = 8; + private static final int ITERATIONS = 20_000; + + private EventLoopGroup group; + private Timer timer; + + @BeforeEach + public void setUp() { + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS); + } + + @AfterEach + public void tearDown() throws Exception { + timer.stop(); + group.shutdownGracefully().get(); + } + + @Test + public void test_concurrentLockUnlock_shouldKeepUnavailableCountConsistent() throws Exception { + IProtoClientPoolImpl.ConnectionStatistics statistics = + new IProtoClientPoolImpl.ConnectionStatistics(); + PoolEntry entry = createEntry(statistics); + + CyclicBarrier start = new CyclicBarrier(THREADS); + List workers = new ArrayList<>(); + for (int t = 0; t < THREADS; t++) { + Thread worker = + new Thread( + () -> { + try { + start.await(); + } catch (Exception e) { + throw new IllegalStateException(e); + } + for (int i = 0; i < ITERATIONS; i++) { + if (i % 2 == 0) { + entry.lock(); + } else { + entry.unlock(); + } + } + }); + workers.add(worker); + worker.start(); + } + for (Thread worker : workers) { + worker.join(); + } + + // The entry counts as unavailable exactly while it is locked; without an atomic flag two + // threads could both pass the check and change the counter twice. + assertEquals(entry.isLocked() ? 1 : 0, statistics.unavailable()); + } + + private PoolEntry createEntry(IProtoClientPoolImpl.ConnectionStatistics statistics) { + Bootstrap bootstrap = new Bootstrap().group(group).channel(NioSocketChannel.class); + InstanceConnectionGroup instanceGroup = + InstanceConnectionGroup.builder() + .withHost(InetAddress.getLoopbackAddress().getHostAddress()) + .withPort(3301) + .withSize(1) + .withTag("lock") + .build(); + return new PoolEntry( + new ConnectionFactory(bootstrap, timer), + timer, + instanceGroup, + 0, + false, + 1000, + 1000, + null, + null, + statistics, + null, + null, + null, + false, + null, + null); + } +} diff --git a/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectDelayTest.java b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectDelayTest.java new file mode 100644 index 00000000..84c86c31 --- /dev/null +++ b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectDelayTest.java @@ -0,0 +1,116 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.pool.unit; + +import java.net.InetAddress; +import java.net.ServerSocket; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.tarantool.core.ManagedResource; +import io.tarantool.core.connection.ConnectionFactory; +import io.tarantool.pool.IProtoClientPoolImpl; +import io.tarantool.pool.InstanceConnectionGroup; +import io.tarantool.pool.PoolEventListener; + +/** Reconnect delays of a pool entry whose server refuses connections. */ +public class ReconnectDelayTest { + + private static final String TAG = "refused"; + + private EventLoopGroup group; + private Timer timer; + private int refusedPort; + private final List delays = Collections.synchronizedList(new ArrayList<>()); + private IProtoClientPoolImpl pool; + + @BeforeEach + public void setUp() throws Exception { + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS); + try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getLoopbackAddress())) { + refusedPort = socket.getLocalPort(); + } + Bootstrap bootstrap = + new Bootstrap() + .group(group) + .channel(NioSocketChannel.class) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000); + PoolEventListener listener = + new PoolEventListener() { + @Override + public void onReconnectScheduled(String tag, int index, long delayMs) { + delays.add(delayMs); + } + }; + pool = + new IProtoClientPoolImpl( + new ConnectionFactory(bootstrap, timer), + ManagedResource.external(timer), + true, + null, + null, + null, + null, + null, + false, + listener); + } + + @AfterEach + public void tearDown() throws Exception { + pool.close(); + timer.stop(); + group.shutdownGracefully().get(); + } + + @Test + public void test_reconnectAfter_shouldApplyToExistingConnections() throws Exception { + // TarantoolClientImpl sets the groups first and the reconnect delay after that. + setGroups(); + pool.setReconnectAfter(120); + pool.get(TAG, 0); + + awaitDelays(1); + + assertEquals(120L, delays.get(0)); + } + + private void setGroups() { + pool.setGroups( + Collections.singletonList( + InstanceConnectionGroup.builder() + .withHost(InetAddress.getLoopbackAddress().getHostAddress()) + .withPort(refusedPort) + .withSize(1) + .withTag(TAG) + .build())); + } + + private void awaitDelays(int count) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10); + while (delays.size() < count && System.nanoTime() < deadline) { + Thread.sleep(10); + } + assertTrue(delays.size() >= count, "reconnects scheduled: " + delays); + } +} diff --git a/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java new file mode 100644 index 00000000..c4779cae --- /dev/null +++ b/tarantool-pooling/src/test/java/io/tarantool/pool/unit/ReconnectLoopTest.java @@ -0,0 +1,227 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.pool.unit; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertTrue; +import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.LongTaskTimer; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.MultiThreadIoEventLoopGroup; +import io.netty.channel.nio.NioIoHandler; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.HashedWheelTimer; +import io.netty.util.Timer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.tarantool.core.ManagedResource; +import io.tarantool.core.connection.ConnectionFactory; +import io.tarantool.pool.IProtoClientPoolImpl; +import io.tarantool.pool.InstanceConnectionGroup; +import io.tarantool.pool.PoolEventListener; + +/** A pool entry keeps reconnecting to a server that accepts connections and never answers. */ +public class ReconnectLoopTest { + + private static final String TAG = "silent"; + private static final long CONNECT_TIMEOUT_MS = 200; + private static final long RECONNECT_AFTER_MS = 50; + private static final long OBSERVE_MS = 3_000; + + private EventLoopGroup group; + private Timer timer; + private ServerSocket server; + private Thread acceptor; + private final List accepted = Collections.synchronizedList(new ArrayList<>()); + private IProtoClientPoolImpl pool; + + @BeforeEach + public void setUp() throws IOException { + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS); + server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + // Accepts connections and never writes the greeting; each accepted socket is one attempt. + acceptor = + new Thread( + () -> { + while (!server.isClosed()) { + try { + accepted.add(server.accept()); + } catch (IOException e) { + return; + } + } + }); + acceptor.setDaemon(true); + acceptor.start(); + } + + @AfterEach + public void tearDown() throws Exception { + if (pool != null) { + pool.close(); + } + server.close(); + synchronized (accepted) { + for (Socket socket : accepted) { + socket.close(); + } + } + timer.stop(); + group.shutdownGracefully().get(); + } + + @Test + public void test_failedAttempts_shouldNotMultiplyReconnectTasks() throws Exception { + AtomicInteger scheduled = new AtomicInteger(); + PoolEventListener listener = + new PoolEventListener() { + @Override + public void onReconnectScheduled(String tag, int index, long delayMs) { + scheduled.incrementAndGet(); + } + }; + pool = createPool(null, listener); + pool.get(TAG, 0); + + Thread.sleep(OBSERVE_MS); + + int attempts = accepted.size(); + assertTrue(attempts >= 3, "too few attempts to judge: " + attempts); + // Each failed attempt schedules one reconnect. When a greeting timeout was also reported as + // a close by server, every attempt was handled twice and the reconnect chains multiplied. + assertTrue( + scheduled.get() <= attempts + 1, + "reconnect tasks multiply: " + + scheduled.get() + + " scheduled for " + + attempts + + " attempts"); + } + + @Test + public void test_closedBeforeGreeting_shouldNotMultiplyReconnectTasks() throws Exception { + AtomicInteger scheduled = new AtomicInteger(); + PoolEventListener listener = + new PoolEventListener() { + @Override + public void onReconnectScheduled(String tag, int index, long delayMs) { + scheduled.incrementAndGet(); + } + }; + AtomicInteger attempts = new AtomicInteger(); + try (ServerSocket closingServer = new ServerSocket(0, 50, InetAddress.getLoopbackAddress())) { + // Accepts and closes at once, before the greeting, like a proxy without a backend. + Thread closer = + new Thread( + () -> { + while (!closingServer.isClosed()) { + try (Socket socket = closingServer.accept()) { + attempts.incrementAndGet(); + } catch (IOException e) { + return; + } + } + }); + closer.setDaemon(true); + closer.start(); + pool = createPool(null, listener, closingServer); + pool.get(TAG, 0); + + Thread.sleep(OBSERVE_MS); + } + + int count = attempts.get(); + assertTrue(count >= 3, "too few attempts to judge: " + count); + // A connection closed before the greeting has never been established: the failed connect + // future is the only report of the attempt, so each attempt schedules one reconnect. + assertTrue( + scheduled.get() <= count + 1, + "reconnect tasks multiply: " + scheduled.get() + " scheduled for " + count + " attempts"); + } + + @Test + public void test_failedAttempts_shouldStopConnectTimer() throws Exception { + MeterRegistry registry = createMetricsRegistry(); + pool = createPool(registry, null); + pool.get(TAG, 0); + + Thread.sleep(OBSERVE_MS); + + int attempts = accepted.size(); + assertTrue(attempts >= 3, "too few attempts to judge: " + attempts); + LongTaskTimer connectTime = registry.find("pool.connect.time").longTaskTimer(); + // At most one attempt of the single connection can be in flight. + assertTrue( + connectTime.activeTasks() <= 1, + "connect timer is not stopped on failure: " + + connectTime.activeTasks() + + " active for " + + attempts + + " attempts"); + } + + private IProtoClientPoolImpl createPool(MeterRegistry registry, PoolEventListener listener) { + return createPool(registry, listener, server); + } + + private IProtoClientPoolImpl createPool( + MeterRegistry registry, PoolEventListener listener, ServerSocket target) { + Bootstrap bootstrap = + new Bootstrap() + .group(group) + .channel(NioSocketChannel.class) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000); + IProtoClientPoolImpl pool = + new IProtoClientPoolImpl( + new ConnectionFactory(bootstrap, timer), + ManagedResource.external(timer), + true, + null, + null, + registry, + null, + null, + false, + listener); + pool.setConnectTimeout(CONNECT_TIMEOUT_MS); + pool.setReconnectAfter(RECONNECT_AFTER_MS); + pool.setGroups( + Collections.singletonList( + InstanceConnectionGroup.builder() + .withHost(target.getInetAddress().getHostAddress()) + .withPort(target.getLocalPort()) + .withSize(1) + .withTag(TAG) + .build())); + return pool; + } + + private static MeterRegistry createMetricsRegistry() { + MeterRegistry registry = new SimpleMeterRegistry(); + LongTaskTimer.builder("request.timer").register(registry); + Counter.builder("request.counter").register(registry); + Counter.builder("response.success").register(registry); + Counter.builder("response.errors").register(registry); + Counter.builder("response.ignored").register(registry); + return registry; + } +} diff --git a/testcontainers/pom.xml b/testcontainers/pom.xml index e6cd35fe..25bbd18b 100644 --- a/testcontainers/pom.xml +++ b/testcontainers/pom.xml @@ -3,7 +3,7 @@ 4.0.0 testcontainers - Minimalistic java connector for Tarantool versions 2.11+ based on Netty framework + Testcontainers for Tarantool: single nodes, vshard and Cartridge clusters, Tarantool DB, TDG, TQE and TCM https://tarantool.io testcontainers