fix: connection level flow control accounting for empty, non-fin frames

[gregor-cf]

This fixes a bug in connection level flow control accounting. When an empty STREAM frame without FIN is received, we calcuate the difference between the highest previously received offset and the frame's max_off and add that to connection level flow control. However, in RecvBuf::write() we return early and we do not update the highest received offset. On subsequent frames, we count the same bytes against conn level flow control again.

To fix this, we update the highest receive offset in RecvBuf::write(). The alternative would be to ignore such frames, but I think updating the high water mark is the right choice: It mimmicks what we do for empty FIN frames and https://datatracker.ietf.org/doc/html/rfc9000#section-19.8 also says

"When a Stream Data field has a length of 0, the offset in the STREAM frame is the offset of the next byte that would be sent."

[antoniovicente]

Fix looks good.

[antoniovicente]

Should this update happen unconditionally before the first return Ok()?

[gregor-cf]

the only return Ok(()) before this line happens when we already have a fin_off in which case the current buffer can't change self.len.

That whole write function could probably be edited and simplified a bit in the long term

[ghedo]

Nit:

Suggested change

	// However, we must still advance the max received offset. A
	//
	// However, we must still advance the max received offset. A

[ghedo]

Nit:

Suggested change

	// NOTE: connection level flow control accounting also expects this.
	//
	// NOTE: connection level flow control accounting also expects this.

[ghedo]

Should self.off also be updated? Specifically if self.drain = true, after stream_shutdown(..., Shutdown::Read, ...), Connection treats the new offset delta as consumed flow control, but off stays stale here. A later RESET_STREAM then computes consumed_flowcontrol = final_size - self.off and can count the same range again.

See following test case which fails on this branch:

#[rstest]
/// Tests that empty non-FIN STREAM frames on a draining stream are not counted
/// again when the same final size is later received in a RESET_STREAM frame.
fn flow_control_drain_empty_stream_frame_reset(
    #[values("cubic", "bbr2_gcongestion")] cc_algorithm_name: &str,
) {
    let mut buf = [0; 65536];

    let mut pipe = test_utils::Pipe::new(cc_algorithm_name).unwrap();
    assert_eq!(pipe.handshake(), Ok(()));

    assert_eq!(pipe.client.stream_send(4, b"aaaaa", false), Ok(5));
    assert_eq!(pipe.advance(), Ok(()));

    assert_eq!(pipe.server.rx_data, 5);
    assert_eq!(pipe.server.flow_control.consumed(), 0);

    assert_eq!(pipe.server.stream_shutdown(4, Shutdown::Read, 42), Ok(()));

    assert_eq!(pipe.server.rx_data, 5);
    assert_eq!(pipe.server.flow_control.consumed(), 5);

    let frames = [frame::Frame::Stream {
        stream_id: 4,
        data: RangeBuf::from(&[], 10, false),
    }];

    let written =
        test_utils::encode_pkt(&mut pipe.client, Type::Short, &frames, &mut buf)
            .unwrap();
    assert_eq!(pipe.server_recv(&mut buf[..written]), Ok(written));

    assert_eq!(pipe.server.rx_data, 10);
    assert_eq!(pipe.server.flow_control.consumed(), 10);

    let frames = [frame::Frame::ResetStream {
        stream_id: 4,
        error_code: 42,
        final_size: 10,
    }];

    let written =
        test_utils::encode_pkt(&mut pipe.client, Type::Short, &frames, &mut buf)
            .unwrap();
    assert_eq!(pipe.server_recv(&mut buf[..written]), Ok(written));

    assert_eq!(pipe.server.rx_data, 10);
    assert_eq!(pipe.server.flow_control.consumed(), 10);
}

So I think we need something like in https://github.com/cloudflare/quiche/blob/master/quiche/src/stream/recv_buf.rs#L194:

Suggested change

	return Ok(());
	if self.drain {
	self.off = self.len;
	}
	return Ok(());