Stop losing outbound channel data when the peer's window fills (claude - testing)

.release-notes/outbound-channel-flow-control.md

@@ -0,0 +1,26 @@
+## Stop losing outbound channel data when the peer's window fills
+
+`SshSession.channel_send` sent as much of a buffer as the peer's flow-control window allowed, dropped the rest, and reported `SshWindowExhausted` through `ssh_channel_error`. That error carried no byte count, so a consumer could not tell how much had gone out or where to resume, and `ssh_channel_error` does nothing unless overridden. A server writing more output to a channel than the peer's window allowed lost the tail of it without a trace.
+
+Data offered past the window is now held and sent as the peer grants more window. A buffer is accepted whole or not at all: when the queue is full `channel_send` sends none of that buffer and reports `SshWindowExhausted`, so a consumer is never left working out which part of its buffer went out. The queue holds up to 256 KiB per channel.
+
+A channel carries a byte stream, so one `channel_send` is not one packet: a buffer may be split across several, and one packet may carry the end of one buffer and the start of the next. Frame your own messages if the peer needs to see boundaries.
+
+`SshClientNotify` and `SshServerNotify` gained `ssh_channel_writeable`, called when a channel whose queue filled has drained empty. It is called only for a channel whose `channel_send` was refused, so a consumer that never overflows never sees it. Both interfaces supply a default that does nothing, so existing implementations still compile.
+
+```pony
+be ssh_channel_error(session: SshSession tag, channel_id: U32,
+  err: SshChannelError val)
+=>
+  match err
+  | SshWindowExhausted => _blocked = true  // none of that buffer was sent
+  end
+
+be ssh_channel_writeable(session: SshSession tag, channel_id: U32) =>
+  if _blocked then
+    _blocked = false
+    session.channel_send(channel_id, _retry_buffer)
+  end
+```
+
+Closing a channel that still has data queued now reports `SshChannelClosed` for the part that was never sent, instead of discarding it silently.

ponyssh/ssh_connection/ssh_channel.pony

@@ -1,3 +1,5 @@
+use "buffered"
+
 class SshChannelState
   let local_id: U32
   var remote_id: U32
@@ -8,6 +10,15 @@ class SshChannelState
   var open: Bool = true
   var pty: (SshPtyState val | None) = None
   var pty_pending: Bool = false
+  // Outbound data accepted from the consumer that the peer's send window has no
+  // room for yet, oldest first, drained as the peer grants window. A Reader
+  // holds it: despite the name it is a queue of byte chunks that hands back a
+  // bounded prefix and keeps the remainder, which is exactly what draining
+  // against a send window needs.
+  embed pending_send: Reader = Reader
+  // Set when a write was refused because the queue was full, so the session
+  // knows to tell the consumer once the queue drains that it may resume.
+  var send_blocked: Bool = false
 
   new create(local_id': U32, remote_id': U32,
     local_window': U32, remote_window': U32,

ponyssh/ssh_connection/ssh_manager.pony

@@ -18,6 +18,16 @@ primitive SshChannelLimits
     """
     256
 
+  fun max_pending_send(): USize =>
+    """
+    Maximum outbound bytes held per channel while waiting on the peer's send
+    window. The queue exists to cover the round trip between exhausting that
+    window and the CHANNEL_WINDOW_ADJUST that reopens it, so it needs to absorb
+    a burst, not a whole transfer. 256 KiB does that while bounding a peer that
+    stops granting window at 64 MiB across the 256-channel cap.
+    """
+    262144
+
 class SshChannelManager
   """
   Tracks channel state keyed by local channel id. The local id (the map key) is
@@ -79,6 +89,86 @@ class SshChannelManager
       SshChannelClosed
     end
 
+  fun ref queue_send(local_id: U32, data: Array[U8] val):
+    (None | SshChannelError)
+  =>
+    """
+    Take outbound channel data, to be drained as the peer's send window allows.
+    All of data is accepted or none of it is: when the queue has no room for the
+    whole buffer the call is rejected with SshWindowExhausted and nothing is
+    queued, so a caller is never left working out how much of its buffer went
+    out. SshChannelClosed means there is no such open channel.
+    """
+    try
+      let ch = _channels(local_id)?
+      if not ch.open then return SshChannelClosed end
+      if data.size() == 0 then return None end
+      if (ch.pending_send.size() + data.size())
+        > SshChannelLimits.max_pending_send()
+      then
+        ch.send_blocked = true
+        return SshWindowExhausted
+      end
+      ch.pending_send.append(data)
+      None
+    else
+      SshChannelClosed
+    end
+
+  fun ref next_send_segment(local_id: U32): (Array[U8] val | None) =>
+    """
+    Take the next segment of queued outbound data that the peer's send window
+    and maximum packet size allow, charging it against the window. None once the
+    queue is empty or the window is full, leaving the remainder to wait for the
+    peer's next CHANNEL_WINDOW_ADJUST.
+    """
+    try
+      let ch = _channels(local_id)?
+      if not ch.open then return None end
+      let queued = ch.pending_send.size()
+      let window = ch.remote_window.usize()
+      if (queued == 0) or (window == 0) then return None end
+      let seg_len = queued.min(_max_packet_size(ch)).min(window)
+      // Charge the window through the one method that owns that accounting.
+      match channel_data_send(local_id, seg_len)
+      | let _: SshChannelError => return None
+      end
+      // seg_len is bounded by what is queued, so this cannot come up short.
+      ch.pending_send.block(seg_len)?
+    else
+      None
+    end
+
+  fun ref take_send_unblocked(local_id: U32): Bool =>
+    """
+    True the once, when a channel whose queue was full has drained empty. The
+    session reports that transition so a consumer whose write was refused knows
+    it may resume; a consumer that never overflowed is never told anything.
+    """
+    try
+      let ch = _channels(local_id)?
+      if ch.send_blocked and (ch.pending_send.size() == 0) then
+        ch.send_blocked = false
+        true
+      else
+        false
+      end
+    else
+      false
+    end
+
+  fun pending_send_bytes(local_id: U32): USize =>
+    """Outbound bytes still queued for a channel; 0 if there is no such channel."""
+    try _channels(local_id)?.pending_send.size() else 0 end
+
+  fun _max_packet_size(ch: SshChannelState box): USize =>
+    """
+    The peer's maximum packet size, or 32 KiB when it advertised none — sending
+    a larger CHANNEL_DATA than the peer allows is rejected by a conformant peer.
+    """
+    let m = ch.max_packet_size.usize()
+    if m == 0 then 32768 else m end
+
   fun ref channel_data_received(local_id: U32, data_size: USize):
     (U32 | SshChannelError)
   =>

ponyssh/ssh_transport/ssh_notify.pony

@@ -18,6 +18,12 @@ interface SshClientNotify
   be ssh_channel_opened(session: SshSession tag, channel_id: U32)
   be ssh_channel_data(session: SshSession tag, channel_id: U32, data: Array[U8] val)
   be ssh_channel_error(session: SshSession tag, channel_id: U32, err: SshChannelError val)
+  be ssh_channel_writeable(session: SshSession tag, channel_id: U32) => None
+    """
+    A channel whose send queue was full has drained. Only sent to a consumer
+    whose channel_send was refused with SshWindowExhausted, telling it the
+    refused data can now be re-sent.
+    """
   be ssh_channel_closed(session: SshSession tag, channel_id: U32)
   be ssh_error(session: SshSession tag, err: SshTransportError val)
   be ssh_disconnected(session: SshSession tag)
@@ -85,6 +91,12 @@ interface SshServerNotify
     data: Array[U8] val) => None
   be ssh_channel_error(session: SshSession tag, channel_id: U32,
     err: SshChannelError val) => None
+  be ssh_channel_writeable(session: SshSession tag, channel_id: U32) => None
+    """
+    A channel whose send queue was full has drained. Only sent to a consumer
+    whose channel_send was refused with SshWindowExhausted, telling it the
+    refused data can now be re-sent.
+    """
   be ssh_channel_closed(session: SshSession tag, channel_id: U32) => None
   be ssh_error(session: SshSession tag, err: SshTransportError val) => None
   be ssh_disconnected(session: SshSession tag) => None

ponyssh/ssh_transport/ssh_session.pony

@@ -161,51 +161,50 @@ actor SshSession
 
   be channel_send(channel_id: U32, data: Array[U8] val) =>
     match _state
-    | let _: SshStateConnected => _channel_send_segmented(channel_id, data)
+    | let _: SshStateConnected => _channel_send(channel_id, data)
     end
 
-  fun ref _channel_send_segmented(channel_id: U32, data: Array[U8] val) =>
+  fun ref _channel_send(channel_id: U32, data: Array[U8] val) =>
     """
-    Split outbound channel data into packets no larger than the peer's
-    advertised max_packet_size (and no larger than its remaining send window),
-    rather than emitting one oversized CHANNEL_DATA a conformant peer rejects.
+    Queue outbound channel data and send as much of it as flow control allows.
+    The buffer is accepted whole or not at all: when the send queue is full the
+    consumer is told with SshWindowExhausted and nothing is sent, so it never
+    has to work out how much of its buffer went out. Anything the peer's window
+    has no room for yet leaves as the peer grants more.
     """
-    let max_packet = match _channel_manager.get(channel_id)
-      | let ch: SshChannelState =>
-        let m = ch.max_packet_size.usize()
-        if m == 0 then 32768 else m end
-      | None =>
-        _notify_channel_error(channel_id, SshChannelClosed)
-        return
-      end
-    var offset: USize = 0
-    while offset < data.size() do
-      let window = match _channel_manager.get(channel_id)
-        | let ch: SshChannelState => ch.remote_window.usize()
-        | None =>
-          _notify_channel_error(channel_id, SshChannelClosed)
-          return
-        end
-      if window == 0 then
-        // The peer's send window is exhausted; the remainder cannot go out
-        // until it grants more via CHANNEL_WINDOW_ADJUST.
-        _notify_channel_error(channel_id, SshWindowExhausted)
-        return
-      end
-      let seg_len = (data.size() - offset).min(max_packet).min(window)
-      let segment = recover val
-        let b = Array[U8].create(seg_len)
-        b.copy_from(data, offset, 0, seg_len)
-        b
+    match _channel_manager.queue_send(channel_id, data)
+    | let err: SshChannelError =>
+      _notify_channel_error(channel_id, err)
+    else
+      _drain_channel(channel_id)
+    end
+
+  fun ref _drain_channel(channel_id: U32) =>
+    """
+    Send as much queued channel data as the peer's send window and maximum
+    packet size allow, splitting it into conformant CHANNEL_DATA packets. Called
+    when data is queued and again whenever the peer grants window; whatever does
+    not fit stays queued.
+
+    An unknown channel is ignored rather than reported. One caller passes an id
+    taken from a peer's CHANNEL_WINDOW_ADJUST, and surfacing that would let a
+    peer raise channel errors for ids that never existed. A consumer's own send
+    is told about a missing channel by queue_send, before this runs.
+    """
+    let remote_id = match _channel_manager.get(channel_id)
+      | let ch: SshChannelState => ch.remote_id
+      | None => return
       end
-      match _channel_manager.channel_data_send(channel_id, seg_len)
-      | let remote_id: U32 =>
+    var draining = true
+    while draining do
+      match _channel_manager.next_send_segment(channel_id)
+      | let segment: Array[U8] val =>
         _send_packet(SshChannelMessages.channel_data(remote_id, segment))
-      | let err: SshChannelError =>
-        _notify_channel_error(channel_id, err)
-        return
+      | None => draining = false
       end
-      offset = offset + seg_len
+    end
+    if _channel_manager.take_send_unblocked(channel_id) then
+      _notify_channel_writeable(channel_id)
     end
 
   be channel_request_shell(channel_id: U32, want_reply: Bool = true) =>
@@ -245,6 +244,11 @@ actor SshSession
     | let _: SshStateConnected =>
       match _channel_manager.channel_data_send(channel_id, 0)
       | let remote_id: U32 =>
+        // Closing drops anything still waiting on the peer's window. Say so
+        // rather than lose it silently.
+        if _channel_manager.pending_send_bytes(channel_id) > 0 then
+          _notify_channel_error(channel_id, SshChannelClosed)
+        end
         _send_packet(SshChannelMessages.channel_eof(remote_id))
         _send_packet(SshChannelMessages.channel_close(remote_id))
         _channel_manager.close_channel(channel_id)
@@ -1552,6 +1556,8 @@ actor SshSession
         let recipient_channel = r.read_u32()?
         let bytes_to_add = r.read_u32()?
         _channel_manager.window_adjust(recipient_channel, bytes_to_add)
+        // The peer has granted more window; send whatever was waiting on it.
+        _drain_channel(recipient_channel)
       end
     | SshChannelMsgTypes.channel_eof() =>
       None
@@ -1812,6 +1818,14 @@ actor SshSession
     | let n: SshServerNotify tag => n.ssh_channel_error(this, channel_id, err)
     end
 
+  fun ref _notify_channel_writeable(channel_id: U32) =>
+    match _client_notify
+    | let n: SshClientNotify tag => n.ssh_channel_writeable(this, channel_id)
+    end
+    match _server_notify
+    | let n: SshServerNotify tag => n.ssh_channel_writeable(this, channel_id)
+    end
+
   fun ref _notify_channel_closed(channel_id: U32) =>
     match _client_notify
     | let n: SshClientNotify tag => n.ssh_channel_closed(this, channel_id)

ponyssh/tests/_tests.pony

@@ -45,6 +45,12 @@ primitive \nodoc\ Tests is TestList
     test(_TestChannelCapacity)
     test(_TestChannelRequestExecEncode)
     test(_TestChannelRequestShellEncode)
+    test(_TestChannelSendSurvivesWindowExhaustion)
+    test(_TestChannelSendQueueAllOrNothing)
+    test(_TestChannelSendSegmentBounds)
+    test(_TestChannelSendCoalescesQueuedWrites)
+    test(_TestChannelSendUnblockedReportedOnce)
+    test(_TestChannelSendQueueRejectsUnknownChannel)
     test(_TestStrictKexPeerAdvertised)
     test(_TestStrictKexMarkerDoesNotWinNegotiation)
     test(_TestPacketResetSequenceNumber)