diff --git a/CHANGELOG.md b/CHANGELOG.md index 716d7705..206f76c8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ exit code, stdout and stderr are logged and included in the error. Before, the error carried only the empty stdout ("Failed to change the app topology via cartridge CLI: "). +### Bug fixes + +- Fixed a request sent while the client closes the connection failing with a netty + `ClosedChannelException` instead of `ConnectionClosedException`: the request passed the state + check, and `close()` closed the channel before the request was written. The root cause is now + `ConnectionClosedException` with the close reason; the netty error is kept as suppressed. + ## [1.7.3] - 2026-10-30 ### Pool 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..5dbb90ee 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 @@ -238,14 +238,30 @@ public CompletableFuture send(IProtoMessage msg) throws IllegalStateExcept return; } - failPromise( - promise, - new ConnectionException(String.format(ERR_SEND_FAILURE, msg), f.cause())); + failPromise(promise, sendFailure(msg, f.cause())); }); return promise; } + /** + * A write that passed the READY check can still fail if the client closes the connection before + * netty writes it. Report it as {@link ConnectionClosedException}, like other requests cut off by + * the close, and keep the netty error as suppressed. + */ + private ConnectionException sendFailure(IProtoMessage msg, Throwable cause) { + String message = String.format(ERR_SEND_FAILURE, msg); + CloseReason reason = closedBy.get(); + if (reason == CloseReason.NO_REASON) { + return new ConnectionException(message, cause); + } + ConnectionClosedException closed = new ConnectionClosedException(reason.getMessage()); + if (cause != null) { + closed.addSuppressed(cause); + } + return new ConnectionException(message, closed); + } + @Override public void close() { log.info("Connection closed by client {}", this); diff --git a/tarantool-core/src/test/java/io/tarantool/core/connection/SendCloseRaceTest.java b/tarantool-core/src/test/java/io/tarantool/core/connection/SendCloseRaceTest.java new file mode 100644 index 00000000..643c8f39 --- /dev/null +++ b/tarantool-core/src/test/java/io/tarantool/core/connection/SendCloseRaceTest.java @@ -0,0 +1,147 @@ +/* + * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY + * All Rights Reserved. + */ + +package io.tarantool.core.connection; + +import java.io.InputStream; +import java.io.OutputStream; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +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.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; + +import io.tarantool.core.connection.exceptions.ConnectionClosedException; +import io.tarantool.core.protocol.requests.IProtoPing; + +/** + * A request passes the READY check, then the client closes the connection before netty writes it. + */ +public class SendCloseRaceTest { + + private static final long WAIT_SECONDS = 5; + + private EventLoopGroup group; + private Timer timer; + private ServerSocket server; + private ConnectionImpl connection; + + @BeforeEach + public void setUp() throws Exception { + // One event loop: a task submitted to the group runs on the connection's channel thread. + group = new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory()); + timer = new HashedWheelTimer(); + server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + Thread acceptor = new Thread(this::serveGreeting, "fake-tarantool"); + acceptor.setDaemon(true); + acceptor.start(); + 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(); + server.close(); + timer.stop(); + group.shutdownGracefully().get(); + } + + @Test + public void test_sendRacingClientClose_shouldFailWithConnectionClosed() throws Exception { + InetSocketAddress address = + new InetSocketAddress(server.getInetAddress(), server.getLocalPort()); + connection.connect(address, 3000).get(WAIT_SECONDS, TimeUnit.SECONDS); + + // Hold the event loop, so the write below is queued behind the close. + CountDownLatch loopHeld = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + group + .next() + .execute( + () -> { + loopHeld.countDown(); + awaitQuietly(release); + connection.close(); + }); + assertTrue(loopHeld.await(WAIT_SECONDS, TimeUnit.SECONDS)); + + CompletableFuture sent = connection.send(new IProtoPing()); + release.countDown(); + + ExecutionException ex = + assertThrows(ExecutionException.class, () -> sent.get(WAIT_SECONDS, TimeUnit.SECONDS)); + Throwable root = ex.getCause(); + while (root.getCause() != null && root.getCause() != root) { + root = root.getCause(); + } + assertInstanceOf(ConnectionClosedException.class, root); + assertEquals("Connection closed by client", root.getMessage()); + } + + private static void awaitQuietly(CountDownLatch latch) { + try { + latch.await(WAIT_SECONDS, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + /** Sends a Tarantool greeting and discards everything the client writes. */ + private void serveGreeting() { + try (Socket socket = server.accept()) { + OutputStream out = socket.getOutputStream(); + out.write(greeting()); + out.flush(); + InputStream in = socket.getInputStream(); + byte[] buf = new byte[1024]; + while (in.read(buf) >= 0) { + // discard + } + } catch (Exception e) { + // the server socket is closed in tearDown + } + } + + private static byte[] greeting() { + String salt = Base64.getEncoder().encodeToString(new byte[32]); + String text = line("Tarantool 3.5.0 (Binary) " + UUID.randomUUID()) + line(salt); + return text.getBytes(StandardCharsets.US_ASCII); + } + + private static String line(String s) { + StringBuilder sb = new StringBuilder(s); + while (sb.length() < 63) { + sb.append(' '); + } + return sb.append('\n').toString(); + } +}