fix(channels): preserve unread data on remote close

sshlib/src/main/kotlin/org/connectbot/sshlib/client/ForwardingChannel.kt

@@ -49,13 +49,17 @@ internal class ForwardingChannel(
 
     private val incomingIngress = Channel<ByteArray>(Channel.UNLIMITED)
     private val _incomingData = Channel<ByteArray>(Channel.RENDEZVOUS)
+
+    @Volatile private var inboundDeliveryOpen = true
     val incomingData: ReceiveChannel<ByteArray> get() = _incomingData
     private val incomingDeliveryJob = connectionScope.launch {
         try {
             for (data in incomingIngress) {
                 _incomingData.send(data)
                 val adjust = window.releaseLocal(data.size)
-                connection.sendWindowAdjust(remoteChannelNumber, adjust)
+                if (inboundDeliveryOpen) {
+                    connection.sendWindowAdjust(remoteChannelNumber, adjust)
+                }
             }
         } finally {
             _incomingData.close()
@@ -90,7 +94,7 @@ internal class ForwardingChannel(
     internal suspend fun onEof() {
         if (!lifecycle.receiveEof {
                 logger.debug("Forwarding channel $localChannelNumber received EOF")
-                incomingIngress.close()
+                finishInboundDelivery()
             }
         ) {
             throw SshException("Received duplicate EOF or EOF after CLOSE on forwarding channel $localChannelNumber")
@@ -100,16 +104,14 @@ internal class ForwardingChannel(
     internal suspend fun onClose() {
         if (!lifecycle.receiveClose { transition ->
                 logger.debug("Forwarding channel $localChannelNumber closed")
+                finishInboundDelivery()
                 if (SshChannelEffect.SEND_CLOSE in transition.effects) {
                     try {
                         connection.sendChannelClose(remoteChannelNumber)
                     } catch (e: Exception) {
                         logger.debug("Failed to send CHANNEL_CLOSE reply", e)
                     }
                 }
-                incomingIngress.close()
-                incomingDeliveryJob.cancel()
-                _incomingData.close()
                 windowAvailable.close()
                 if (SshChannelEffect.CLOSE_CHANNEL in transition.effects) {
                     connection.notifyChannelClosed(localChannelNumber)
@@ -122,9 +124,7 @@ internal class ForwardingChannel(
 
     internal suspend fun onDisconnected() {
         lifecycle.disconnect { transition ->
-            incomingIngress.close()
-            incomingDeliveryJob.cancel()
-            _incomingData.close()
+            abortInboundDelivery()
             windowAvailable.close()
             if (SshChannelEffect.CLOSE_CHANNEL in transition.effects) {
                 connection.notifyChannelClosed(localChannelNumber)
@@ -134,6 +134,17 @@ internal class ForwardingChannel(
 
     internal suspend fun receiveRequest(action: suspend () -> Unit): Boolean = lifecycle.receiveRequest { action() }
 
+    private fun finishInboundDelivery() {
+        inboundDeliveryOpen = false
+        incomingIngress.close()
+    }
+
+    private fun abortInboundDelivery() {
+        finishInboundDelivery()
+        incomingDeliveryJob.cancel()
+        _incomingData.close()
+    }
+
     suspend fun sendData(data: ByteArray) {
         var offset = 0
         while (offset < data.size) {
@@ -159,13 +170,16 @@ internal class ForwardingChannel(
 
     suspend fun close() {
         lifecycle.sendClose { transition ->
-            incomingIngress.close()
-            incomingDeliveryJob.cancel()
-            _incomingData.close()
+            abortInboundDelivery()
+            windowAvailable.close()
             connection.sendChannelClose(remoteChannelNumber)
             if (SshChannelEffect.CLOSE_CHANNEL in transition.effects) {
                 connection.notifyChannelClosed(localChannelNumber)
             }
         }
+        // A remote CLOSE makes sendClose a no-op, but close() still owns
+        // releasing any unread delivery job retained for graceful draining.
+        abortInboundDelivery()
+        windowAvailable.close()
     }
 }

sshlib/src/main/kotlin/org/connectbot/sshlib/client/SessionChannel.kt

@@ -71,6 +71,8 @@ class SessionChannel internal constructor(
     private val _stderr = Channel<ByteArray>(Channel.RENDEZVOUS)
     private val _extendedData = Channel<Pair<Int, ByteArray>>(Channel.RENDEZVOUS)
 
+    @Volatile private var inboundDeliveryOpen = true
+
     private val stdoutDeliveryJob = connectionScope.launch {
         deliverData(stdoutIngress, _stdout) { it.size }
     }
@@ -150,7 +152,9 @@ class SessionChannel internal constructor(
             for (value in ingress) {
                 output.send(value)
                 val adjust = window.releaseLocal(sizeOf(value))
-                connection.sendWindowAdjust(_remoteChannelNumber, adjust)
+                if (inboundDeliveryOpen) {
+                    connection.sendWindowAdjust(_remoteChannelNumber, adjust)
+                }
             }
         } finally {
             output.close()
@@ -171,9 +175,7 @@ class SessionChannel internal constructor(
     internal suspend fun onEof() {
         if (!lifecycle.receiveEof {
                 logger.debug("Received EOF on channel $localChannelNumber")
-                stdoutIngress.close()
-                stderrIngress.close()
-                extendedDataIngress.close()
+                finishInboundDelivery()
             }
         ) {
             throw org.connectbot.sshlib.SshException("Received duplicate EOF or EOF after CLOSE on channel $localChannelNumber")
@@ -182,7 +184,11 @@ class SessionChannel internal constructor(
 
     internal suspend fun onClose() {
         if (!lifecycle.receiveClose { transition ->
-                closeResources(SshChannelEffect.SEND_CLOSE in transition.effects, "Received CLOSE")
+                closeResources(
+                    replyRequired = SshChannelEffect.SEND_CLOSE in transition.effects,
+                    preserveInbound = SshChannelEffect.CLOSE_INBOUND_STREAMS in transition.effects,
+                    reason = "Received CLOSE",
+                )
                 if (SshChannelEffect.CLOSE_CHANNEL in transition.effects) {
                     connection.notifyChannelClosed(localChannelNumber)
                 }
@@ -192,28 +198,41 @@ class SessionChannel internal constructor(
         }
     }
 
-    private suspend fun closeResources(replyRequired: Boolean, reason: String) {
+    private fun finishInboundDelivery() {
+        inboundDeliveryOpen = false
+        stdoutIngress.close()
+        stderrIngress.close()
+        extendedDataIngress.close()
+    }
+
+    private fun abortInboundDelivery() {
+        finishInboundDelivery()
+        stdoutDeliveryJob.cancel()
+        stderrDeliveryJob.cancel()
+        extendedDeliveryJob.cancel()
+        _stdout.close()
+        _stderr.close()
+        _extendedData.close()
+    }
+
+    private suspend fun closeResources(replyRequired: Boolean, preserveInbound: Boolean, reason: String) {
         logger.debug("$reason on channel $localChannelNumber")
         obfuscatorMutex.withLock {
             obfuscator?.stop()
         }
         chaffJob?.cancel()
+        if (preserveInbound) {
+            finishInboundDelivery()
+        } else {
+            abortInboundDelivery()
+        }
         if (replyRequired) {
             try {
                 connection.sendChannelClose(_remoteChannelNumber)
             } catch (e: Exception) {
                 logger.debug("Failed to send CHANNEL_CLOSE reply", e)
             }
         }
-        stdoutIngress.close()
-        stderrIngress.close()
-        extendedDataIngress.close()
-        stdoutDeliveryJob.cancel()
-        stderrDeliveryJob.cancel()
-        extendedDeliveryJob.cancel()
-        _stdout.close()
-        _stderr.close()
-        _extendedData.close()
         windowAvailable.close()
         // Channel is gone; if the server never reported an exit, resolve
         // waiters with "unknown" rather than leaving them suspended.
@@ -222,7 +241,7 @@ class SessionChannel internal constructor(
 
     internal suspend fun onDisconnected() {
         lifecycle.disconnect { transition ->
-            closeResources(replyRequired = false, reason = "Disconnected")
+            closeResources(replyRequired = false, preserveInbound = false, reason = "Disconnected")
             if (SshChannelEffect.CLOSE_CHANNEL in transition.effects) {
                 connection.notifyChannelClosed(localChannelNumber)
             }
@@ -461,15 +480,8 @@ class SessionChannel internal constructor(
                 logger.debug("Closing channel $localChannelNumber")
                 obfuscatorMutex.withLock { obfuscator?.stop() }
                 chaffJob?.cancel()
-                stdoutIngress.close()
-                stderrIngress.close()
-                extendedDataIngress.close()
-                stdoutDeliveryJob.cancel()
-                stderrDeliveryJob.cancel()
-                extendedDeliveryJob.cancel()
-                _stdout.close()
-                _stderr.close()
-                _extendedData.close()
+                abortInboundDelivery()
+                windowAvailable.close()
                 _exitInfo.complete(null)
                 try {
                     connection.sendChannelClose(_remoteChannelNumber)
@@ -480,6 +492,11 @@ class SessionChannel internal constructor(
                     connection.notifyChannelClosed(localChannelNumber)
                 }
             }
+            // A remote CLOSE makes sendClose a no-op, but close() still owns
+            // releasing any unread delivery job retained for graceful draining.
+            abortInboundDelivery()
+            windowAvailable.close()
+            _exitInfo.complete(null)
         }
     }
 }

sshlib/src/main/kotlin/org/connectbot/sshlib/protocol/SshChannelStateMachine.kt

@@ -308,14 +308,18 @@ internal class SshChannelStateMachine(
         class SendClose(action: suspend (SshChannelAcceptedTransition) -> Unit) :
             ChannelEvent(
                 SshChannelEventId.SEND_CLOSE,
-                setOf(SshChannelEffect.SEND_CLOSE),
+                setOf(SshChannelEffect.SEND_CLOSE, SshChannelEffect.CLOSE_INBOUND_STREAMS),
                 SshChannelEventOrigin.LOCAL_COMMAND,
                 action,
             )
         class ReceiveClose(action: suspend (SshChannelAcceptedTransition) -> Unit) :
             ChannelEvent(
                 SshChannelEventId.RECEIVE_CLOSE,
-                setOf(SshChannelEffect.SEND_CLOSE, SshChannelEffect.CLOSE_CHANNEL),
+                setOf(
+                    SshChannelEffect.SEND_CLOSE,
+                    SshChannelEffect.CLOSE_INBOUND_STREAMS,
+                    SshChannelEffect.CLOSE_CHANNEL,
+                ),
                 SshChannelEventOrigin.PARSED_PACKET,
                 action,
             )

sshlib/src/test/kotlin/org/connectbot/sshlib/client/FakeSshServer.kt

@@ -45,6 +45,7 @@ import org.connectbot.sshlib.protocol.SshMsgChannelOpenConfirmation
 import org.connectbot.sshlib.protocol.SshMsgChannelOpenFailure
 import org.connectbot.sshlib.protocol.SshMsgChannelRequest
 import org.connectbot.sshlib.protocol.SshMsgChannelSuccess
+import org.connectbot.sshlib.protocol.SshMsgChannelWindowAdjust
 import org.connectbot.sshlib.protocol.SshMsgDisconnect
 import org.connectbot.sshlib.protocol.SshMsgExtInfo
 import org.connectbot.sshlib.protocol.SshMsgIgnore
@@ -118,6 +119,7 @@ class FakeSshServer(
     private val receivedChannelOpenConfirmations = Channel<SshMsgChannelOpenConfirmation>(Channel.UNLIMITED)
     private val receivedChannelOpenFailures = Channel<SshMsgChannelOpenFailure>(Channel.UNLIMITED)
     private val receivedChannelData = Channel<SshMsgChannelData>(Channel.UNLIMITED)
+    private val receivedChannelWindowAdjusts = Channel<SshMsgChannelWindowAdjust>(Channel.UNLIMITED)
     private val receivedUnimplemented = Channel<SshMsgUnimplemented>(Channel.UNLIMITED)
 
     fun start(ignoreTransportErrors: Boolean = false) {
@@ -270,6 +272,13 @@ class FakeSshServer(
                             receivedChannelData.trySend(data)
                         }
 
+                        SshEnums.MessageType.SSH_MSG_CHANNEL_WINDOW_ADJUST -> {
+                            val bodyBytes = rawBytes.copyOfRange(1, rawBytes.size)
+                            val adjust = SshMsgChannelWindowAdjust(ByteBufferKaitaiStream(bodyBytes))
+                            adjust._read()
+                            receivedChannelWindowAdjusts.trySend(adjust)
+                        }
+
                         SshEnums.MessageType.SSH_MSG_PING -> {
                             val bodyBytes = rawBytes.copyOfRange(1, rawBytes.size)
                             val pingMsg = SshMsgPing(ByteBufferKaitaiStream(bodyBytes))
@@ -952,4 +961,6 @@ class FakeSshServer(
     suspend fun awaitChannelOpenFailure(): SshMsgChannelOpenFailure = receivedChannelOpenFailures.receive()
 
     suspend fun awaitChannelData(): SshMsgChannelData = receivedChannelData.receive()
+
+    suspend fun awaitChannelWindowAdjust(): SshMsgChannelWindowAdjust = receivedChannelWindowAdjusts.receive()
 }

sshlib/src/test/kotlin/org/connectbot/sshlib/client/ForwardingChannelTest.kt

@@ -108,6 +108,33 @@ class ForwardingChannelTest {
         assertTrue(channel.incomingData.isClosedForReceive)
     }
 
+    @Test
+    fun `remote close preserves unread incoming data until consumed`() = runTest {
+        val (channel, conn) = createChannel()
+        val first = "first".toByteArray()
+        val second = "second".toByteArray()
+
+        channel.onData(first)
+        channel.onData(second)
+        channel.onClose()
+
+        assertArrayEquals(first, channel.incomingData.receive())
+        assertArrayEquals(second, channel.incomingData.receive())
+        assertTrue(channel.incomingData.receiveCatching().isClosed)
+        coVerify(exactly = 0) { conn.sendWindowAdjust(any(), any()) }
+    }
+
+    @Test
+    fun `explicit close after remote close discards abandoned incoming data`() = runTest {
+        val (channel, _) = createChannel()
+
+        channel.onData("abandoned".toByteArray())
+        channel.onClose()
+        channel.close()
+
+        assertTrue(channel.incomingData.receiveCatching().isClosed)
+    }
+
     @Test
     fun `sendData chunks data by maxPacketSize`() = runTest {
         val conn = mockk<SshConnection>(relaxed = true)

sshlib/src/test/kotlin/org/connectbot/sshlib/client/SessionChannelTest.kt

@@ -153,6 +153,63 @@ class SessionChannelTest {
         assertArrayEquals(testData, received)
     }
 
+    @Test
+    fun `remote close preserves unread stdout until consumed`() = runTest {
+        val (channel, conn) = createChannel()
+        val first = "first".toByteArray()
+        val second = "second".toByteArray()
+
+        channel.onData(first)
+        channel.onData(second)
+        channel.onClose()
+
+        assertArrayEquals(first, channel.stdout.receive())
+        assertArrayEquals(second, channel.stdout.receive())
+        assertTrue(channel.stdout.receiveCatching().isClosed)
+        coVerify(exactly = 0) { conn.sendWindowAdjust(any(), any()) }
+    }
+
+    @Test
+    fun `remote close preserves unread stderr and extended data`() = runTest {
+        val (channel, _) = createChannel()
+        val stderr = "error".toByteArray()
+        val extended = "extended".toByteArray()
+
+        channel.onExtendedData(1, stderr)
+        channel.onExtendedData(7, extended)
+        channel.onClose()
+
+        assertArrayEquals(stderr, channel.stderr.receive())
+        val receivedExtended = channel.readExtended()
+        assertEquals(7, receivedExtended?.first)
+        assertArrayEquals(extended, receivedExtended?.second)
+        assertTrue(channel.stderr.receiveCatching().isClosed)
+        assertEquals(null, channel.readExtended())
+    }
+
+    @Test
+    fun `remote EOF preserves unread stdout until consumed`() = runTest {
+        val (channel, _) = createChannel()
+        val data = "tail".toByteArray()
+
+        channel.onData(data)
+        channel.onEof()
+
+        assertArrayEquals(data, channel.stdout.receive())
+        assertTrue(channel.stdout.receiveCatching().isClosed)
+    }
+
+    @Test
+    fun `explicit close after remote close discards abandoned output`() = runTest {
+        val (channel, _) = createChannel()
+
+        channel.onData("abandoned".toByteArray())
+        channel.onClose()
+        channel.close()
+
+        assertTrue(channel.stdout.receiveCatching().isClosed)
+    }
+
     @Test
     fun `close marks channel not open and sends channel close`() = runTest {
         val (channel, conn) = createChannel()