diff --git a/.release-notes/outbound-channel-flow-control.md b/.release-notes/outbound-channel-flow-control.md new file mode 100644 index 0000000..110704f --- /dev/null +++ b/.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. diff --git a/ponyssh/ssh_connection/ssh_channel.pony b/ponyssh/ssh_connection/ssh_channel.pony index e63b161..24dcec8 100644 --- a/ponyssh/ssh_connection/ssh_channel.pony +++ b/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, diff --git a/ponyssh/ssh_connection/ssh_manager.pony b/ponyssh/ssh_connection/ssh_manager.pony index 9237844..c842682 100644 --- a/ponyssh/ssh_connection/ssh_manager.pony +++ b/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) => diff --git a/ponyssh/ssh_transport/ssh_notify.pony b/ponyssh/ssh_transport/ssh_notify.pony index d3f02fa..3112709 100644 --- a/ponyssh/ssh_transport/ssh_notify.pony +++ b/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 diff --git a/ponyssh/ssh_transport/ssh_session.pony b/ponyssh/ssh_transport/ssh_session.pony index ba51035..0a0c903 100644 --- a/ponyssh/ssh_transport/ssh_session.pony +++ b/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) diff --git a/ponyssh/tests/_tests.pony b/ponyssh/tests/_tests.pony index 98c98a8..a9f320f 100644 --- a/ponyssh/tests/_tests.pony +++ b/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) diff --git a/ponyssh/tests/ssh_channel_test.pony b/ponyssh/tests/ssh_channel_test.pony index 227c7bc..12ca330 100644 --- a/ponyssh/tests/ssh_channel_test.pony +++ b/ponyssh/tests/ssh_channel_test.pony @@ -121,6 +121,232 @@ class iso _TestChannelCapacity is UnitTest h.assert_eq[USize](SshChannelLimits.max_concurrent(), mgr.channel_count()) h.assert_true(mgr.at_capacity()) +primitive _ChannelBytes + fun apply(size: USize): Array[U8] val => + """ + A byte pattern whose value changes with position, so a test that reassembles + it detects bytes delivered out of order, not just the right count. + """ + recover val + let a = Array[U8](size) + var i: USize = 0 + while i < size do + a.push((i % 251).u8()) + i = i + 1 + end + a + end + +primitive _ChannelDrain + fun segments(mgr: SshChannelManager ref, id: U32): Array[Array[U8] val] => + """Every segment the manager will release right now, in order.""" + let out = Array[Array[U8] val] + var draining = true + while draining do + match mgr.next_send_segment(id) + | let seg: Array[U8] val => out.push(seg) + | None => draining = false + end + end + out + + fun joined(mgr: SshChannelManager ref, id: U32): Array[U8] val => + """Those segments concatenated, as the peer would see them.""" + let out = recover iso Array[U8] end + for seg in segments(mgr, id).values() do + out.append(seg) + end + consume out + +class iso _TestChannelSendSurvivesWindowExhaustion is UnitTest + """ + Data offered beyond the peer's send window is held and delivered once the peer + grants more, byte for byte and in order. Before the send queue existed the + tail past the window was dropped and the caller was told only that the window + was exhausted, with no way to learn how much had gone out. + """ + fun name(): String => "ssh_channel/send_survives_window_exhaustion" + + fun apply(h: TestHelper) => + let mgr: SshChannelManager ref = SshChannelManager + let id = mgr.open_channel("session") + // A 100-byte send window and 32-byte packets: 250 bytes cannot go out in + // one pass, so the queue has to hold the rest. + mgr.confirm_channel(id, 4, 100, 32) + + let data = _ChannelBytes(250) + match mgr.queue_send(id, data) + | let e: SshChannelError => h.fail("queue must accept: " + e.string()) + end + + // Only what the window allows leaves now; the rest stays queued. + let first = _ChannelDrain.joined(mgr, id) + h.assert_eq[USize](100, first.size()) + h.assert_eq[USize](150, mgr.pending_send_bytes(id)) + + mgr.window_adjust(id, 1000) + let second = _ChannelDrain.joined(mgr, id) + h.assert_eq[USize](150, second.size()) + h.assert_eq[USize](0, mgr.pending_send_bytes(id)) + + let delivered = recover iso Array[U8] end + delivered.append(first) + delivered.append(second) + let all: Array[U8] val = consume delivered + h.assert_array_eq[U8](data, all) + +class iso _TestChannelSendQueueAllOrNothing is UnitTest + """ + A write the queue cannot hold whole is refused outright and nothing is queued, + so a caller is never left working out which part of its buffer went out. + """ + fun name(): String => "ssh_channel/send_queue_all_or_nothing" + + fun apply(h: TestHelper) => + let mgr: SshChannelManager ref = SshChannelManager + let id = mgr.open_channel("session") + // No send window, so everything offered stays queued. + mgr.confirm_channel(id, 4, 0, 32) + + let cap = SshChannelLimits.max_pending_send() + match mgr.queue_send(id, _ChannelBytes(cap)) + | let e: SshChannelError => h.fail("a write that fills the queue must fit") + end + h.assert_eq[USize](cap, mgr.pending_send_bytes(id)) + + match mgr.queue_send(id, _ChannelBytes(1)) + | SshWindowExhausted => None + | let e: SshChannelError => h.fail("expected SshWindowExhausted") + | None => h.fail("a write past the cap must be refused") + end + // Refused whole: the byte over the cap left no trace. + h.assert_eq[USize](cap, mgr.pending_send_bytes(id)) + +class iso _TestChannelSendSegmentBounds is UnitTest + """ + No segment exceeds the peer's maximum packet size โ€” a conformant peer rejects + an oversized CHANNEL_DATA โ€” and the bytes reassemble exactly whatever the + relationship between the queued length and that size. + """ + fun name(): String => "ssh_channel/send_segment_bounds" + + fun apply(h: TestHelper) => + // Below one packet, exactly one, an exact multiple, and a ragged remainder. + let cases = [as (USize, USize): (1, 16); (16, 16); (64, 16); (65, 16)] + for (total, max_packet) in cases.values() do + let mgr: SshChannelManager ref = SshChannelManager + let id = mgr.open_channel("session") + mgr.confirm_channel(id, 4, 100000, max_packet.u32()) + + let data = _ChannelBytes(total) + mgr.queue_send(id, data) + + let segments = _ChannelDrain.segments(mgr, id) + let joined = recover iso Array[U8] end + for seg in segments.values() do + h.assert_true(seg.size() <= max_packet, + "segment of " + seg.size().string() + " exceeds max packet " + + max_packet.string()) + joined.append(seg) + end + let all: Array[U8] val = consume joined + h.assert_array_eq[U8](data, all) + end + +class iso _TestChannelSendCoalescesQueuedWrites is UnitTest + """ + Separate writes are a byte stream, not packet boundaries: one segment may + carry the tail of one write and the head of the next, so a caller's framing + is never assumed to survive into CHANNEL_DATA packets. + """ + fun name(): String => "ssh_channel/send_coalesces_writes" + + fun apply(h: TestHelper) => + let mgr: SshChannelManager ref = SshChannelManager + let id = mgr.open_channel("session") + // 16-byte packets against two 10-byte writes: the first segment can only be + // full if it draws from both. + mgr.confirm_channel(id, 4, 100000, 16) + + let first = _ChannelBytes(10) + let second = _ChannelBytes(10) + mgr.queue_send(id, first) + mgr.queue_send(id, second) + + let segments = _ChannelDrain.segments(mgr, id) + h.assert_eq[USize](2, segments.size()) + try + h.assert_eq[USize](16, segments(0)?.size()) + h.assert_eq[USize](4, segments(1)?.size()) + else + h.fail("expected two segments") + end + + let expected = recover iso Array[U8] end + expected.append(first) + expected.append(second) + let joined = recover iso Array[U8] end + for seg in segments.values() do + joined.append(seg) + end + h.assert_array_eq[U8](consume expected, consume joined) + +class iso _TestChannelSendUnblockedReportedOnce is UnitTest + """ + The drained-empty signal is raised only for a channel whose write was refused, + and only once, so a consumer that never overflowed is never woken and one that + did is not woken repeatedly. + """ + fun name(): String => "ssh_channel/send_unblocked_once" + + fun apply(h: TestHelper) => + let mgr: SshChannelManager ref = SshChannelManager + let cap = SshChannelLimits.max_pending_send() + + // A channel that queues and drains without ever being refused. + let quiet = mgr.open_channel("session") + mgr.confirm_channel(quiet, 4, 0, 32) + mgr.queue_send(quiet, _ChannelBytes(64)) + mgr.window_adjust(quiet, 1000) + _ChannelDrain.joined(mgr, quiet) + h.assert_eq[USize](0, mgr.pending_send_bytes(quiet)) + h.assert_false(mgr.take_send_unblocked(quiet)) + + // A channel that fills its queue, is refused, then drains empty. + let noisy = mgr.open_channel("session") + mgr.confirm_channel(noisy, 5, 0, 32) + mgr.queue_send(noisy, _ChannelBytes(cap)) + match mgr.queue_send(noisy, _ChannelBytes(1)) + | SshWindowExhausted => None + else + h.fail("a write past the cap must be refused") + end + + // Still queued: nothing to report until it has actually drained. + h.assert_false(mgr.take_send_unblocked(noisy)) + + mgr.window_adjust(noisy, cap.u32()) + _ChannelDrain.joined(mgr, noisy) + h.assert_eq[USize](0, mgr.pending_send_bytes(noisy)) + h.assert_true(mgr.take_send_unblocked(noisy)) + // Reported once and cleared, so the consumer is not woken again. + h.assert_false(mgr.take_send_unblocked(noisy)) + +class iso _TestChannelSendQueueRejectsUnknownChannel is UnitTest + """ + Queuing for a channel that does not exist is a closed-channel error, not a + silent discard. + """ + fun name(): String => "ssh_channel/send_queue_unknown_channel" + + fun apply(h: TestHelper) => + let mgr: SshChannelManager ref = SshChannelManager + match mgr.queue_send(99, _ChannelBytes(4)) + | SshChannelClosed => None + | let e: SshChannelError => h.fail("expected SshChannelClosed") + | None => h.fail("queuing for an unknown channel must fail") + end + class iso _TestChannelRequestExecEncode is UnitTest """ The exec channel-request encoder lays out the exact RFC 4254 ยง6.5 wire