Make socket traffic shutdown idempotent

Treat ENOTCONN and WSAENOTCONN as successful traffic shutdown because the peer may finish a graceful close before the owner cancels the I/O thread. Preserve all other failures and descriptor ownership.

Add real loopback coverage for peer-first close and a synthetic failure case that verifies non-benign shutdown errors remain visible.

Author: bluestreak01

core/src/main/c/share/net.c

@@ -130,7 +130,8 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_send
 
 JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_shutdown
         (JNIEnv *e, jclass cl, jint fd) {
-    return shutdown((int) fd, SHUT_RDWR);
+    const int result = shutdown((int) fd, SHUT_RDWR);
+    return result == -1 && errno == ENOTCONN ? 0 : result;
 }
 
 JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_recv

core/src/main/c/windows/net.c

@@ -234,6 +234,11 @@ JNIEXPORT jint JNICALL Java_io_questdb_client_network_Net_shutdown
         (JNIEnv *e, jclass cl, jint fd) {
     const int result = shutdown((SOCKET) fd, SD_BOTH);
     if (result == SOCKET_ERROR) {
+        const int error = WSAGetLastError();
+        if (error == WSAENOTCONN) {
+            return 0;
+        }
+        WSASetLastError(error);
         SaveLastError();
     }
     return result;

core/src/test/java/io/questdb/client/test/network/SocketTrafficShutdownTest.java

@@ -259,6 +259,84 @@ public void testCompatibilityDefaultsDoNotBypassCustomTransportOwnership() {
                 0, socketCloseCount.get());
     }
 
+    @Test(timeout = 30_000L)
+    public void testPlainSocketShutdownAfterPeerDisconnectRetainsFd() throws Exception {
+        Socket socket = new PlainSocket(NF, LoggerFactory.getLogger(SocketTrafficShutdownTest.class));
+
+        long buffer = 0;
+        int fd = -1;
+        try (ServerSocket listener = new ServerSocket()) {
+            listener.bind(new InetSocketAddress("127.0.0.1", 0));
+            long addrInfo = NF.getAddrInfo("127.0.0.1", listener.getLocalPort());
+            Assert.assertNotEquals(-1L, addrInfo);
+            try {
+                fd = NF.socketTcp(true);
+                Assert.assertTrue("could not allocate client socket", fd >= 0);
+                Assert.assertEquals(0, NF.connectAddrInfo(fd, addrInfo));
+            } finally {
+                NF.freeAddrInfo(addrInfo);
+            }
+
+            try (java.net.Socket peer = listener.accept()) {
+                socket.of(fd);
+                int retainedFd = fd;
+                fd = -1;
+                buffer = Unsafe.malloc(1, MemoryTag.NATIVE_DEFAULT);
+
+                peer.close();
+                Assert.assertTrue("client must observe the peer disconnect", socket.recv(buffer, 1) < 0);
+
+                socket.closeTraffic();
+                Assert.assertEquals("traffic cancellation must retain fd ownership", retainedFd, socket.getFd());
+                Assert.assertFalse("traffic cancellation must not perform full close", socket.isClosed());
+                Assert.assertTrue("shutdown must leave the descriptor allocated", NF.getSndBuf(retainedFd) > 0);
+
+                socket.close();
+                Assert.assertTrue("full close must release the retained fd", socket.isClosed());
+                Assert.assertEquals("released descriptor must reject socket operations", -1, NF.getSndBuf(retainedFd));
+            }
+        } finally {
+            socket.close();
+            if (buffer != 0) {
+                Unsafe.free(buffer, 1, MemoryTag.NATIVE_DEFAULT);
+            }
+            if (fd != -1) {
+                NF.close(fd);
+            }
+        }
+    }
+
+    @Test
+    public void testPlainSocketShutdownFailureStillThrows() {
+        AtomicInteger closeCount = new AtomicInteger();
+        NetworkFacade failingFacade = new CompatibilityNetworkFacade(closeCount) {
+            @Override
+            public int errno() {
+                return 1234;
+            }
+
+            @Override
+            public int shutdown(int fd) {
+                Assert.assertEquals(42, fd);
+                return -1;
+            }
+        };
+        PlainSocket socket = new PlainSocket(failingFacade, LoggerFactory.getLogger(SocketTrafficShutdownTest.class));
+        socket.of(42);
+
+        try {
+            socket.closeTraffic();
+            Assert.fail("expected genuine traffic shutdown failure");
+        } catch (IllegalStateException expected) {
+            Assert.assertEquals("could not shut down socket traffic [fd=42, errno=1234]", expected.getMessage());
+        } finally {
+            socket.close();
+        }
+
+        Assert.assertTrue(socket.isClosed());
+        Assert.assertEquals("full close must release facade ownership exactly once", 1, closeCount.get());
+    }
+
     @Test(timeout = 30_000L)
     public void testPlainSocketShutdownWakesMacOsKqueueAndRetainsFd() throws Exception {
         assertShutdownWakesMacOsKqueue(new PlainSocket(NF, LoggerFactory.getLogger(SocketTrafficShutdownTest.class)));