quic

%%% -*- erlang -*-
%%%
%%% QUIC Flow Control
%%% RFC 9000 Section 4 - Flow Control
%%%
%%% Copyright (c) 2024-2026 Benoit Chesneau
%%% Apache License 2.0
%%%
%%% @doc QUIC connection-level flow control implementation.
%%%
%%% This module manages flow control at the connection level:
%%% - Tracking bytes sent against peer's MAX_DATA limit
%%% - Tracking bytes received against our MAX_DATA limit
%%% - Generating MAX_DATA frames when needed
%%% - Detecting when we're blocked by flow control
%%%
%%% == Flow Control Concepts ==
%%%
%%% - MAX_DATA: Maximum total bytes the peer can send
%%% - DATA_BLOCKED: Indicates sender is blocked by receiver's limit
%%% - Window: The difference between max_data and bytes received
%%%

-module(quic_flow).

-include("quic.hrl").

-export([
    %% State management
    new/0,
    new/1,

    %% Send side
    can_send/2,
    on_data_sent/2,
    on_max_data_received/2,
    send_blocked/1,

    %% Receive side
    on_data_received/2,
    should_send_max_data/1,
    generate_max_data/1,

    %% Queries
    bytes_sent/1,
    bytes_received/1,
    send_limit/1,
    recv_limit/1,
    send_window/1,
    recv_window/1
]).

%% Default values

%% Threshold encodes "fraction of window REMAINING when MAX_DATA fires":
%% fires when bytes_received > recv_max_data_sent - (Initial * THRESHOLD).
%% So 0.5 fires after 50% consumed, 0.75 fires after 25% consumed.
%%
%% Bumped from 0.5 to 0.75 so MAX_DATA updates arrive twice as often
%% under sustained high-throughput. The previous 0.5 threshold stalled
%% sustained single-stream transfers because the sender would deplete
%% the remaining half of the window before the next MAX_DATA frame
%% arrived.
-define(WINDOW_UPDATE_THRESHOLD, 0.75).

%% Flow control state
-record(flow_state, {
    %% Send side (our sending, limited by peer's MAX_DATA)
    bytes_sent = 0 :: non_neg_integer(),
    % Peer's limit on what we can send
    send_max_data :: non_neg_integer(),
    send_blocked = false :: boolean(),

    %% Receive side (peer's sending, limited by our MAX_DATA)
    bytes_received = 0 :: non_neg_integer(),
    % Our limit on what peer can send
    recv_max_data :: non_neg_integer(),
    % Last MAX_DATA we sent
    recv_max_data_sent :: non_neg_integer(),

    %% Configuration
    initial_max_data :: non_neg_integer()
}).

-opaque flow_state() :: #flow_state{}.
-export_type([flow_state/0]).

%%====================================================================
%% State Management
%%====================================================================

%% @doc Create a new flow control state.
-spec new() -> flow_state().
new() ->
    new(#{}).

%% @doc Create a new flow control state with options.
-spec new(map()) -> flow_state().
new(Opts) ->
    InitialMaxData = maps:get(initial_max_data, Opts, ?DEFAULT_INITIAL_MAX_DATA),
    PeerMaxData = maps:get(peer_initial_max_data, Opts, ?DEFAULT_INITIAL_MAX_DATA),
    #flow_state{
        send_max_data = PeerMaxData,
        recv_max_data = InitialMaxData,
        recv_max_data_sent = InitialMaxData,
        initial_max_data = InitialMaxData
    }.

%%====================================================================
%% Send Side
%%====================================================================

%% @doc Check if we can send the specified number of bytes.
-spec can_send(flow_state(), non_neg_integer()) -> boolean().
can_send(#flow_state{bytes_sent = Sent, send_max_data = Max}, Size) ->
    Sent + Size =< Max.

%% @doc Record that we sent data.
%% Returns {ok, NewState} or {blocked, NewState} if we hit the limit.
-spec on_data_sent(flow_state(), non_neg_integer()) ->
    {ok | blocked, flow_state()}.
on_data_sent(#flow_state{bytes_sent = Sent, send_max_data = Max} = State, Size) ->
    NewSent = Sent + Size,
    NewState = State#flow_state{bytes_sent = NewSent},
    case NewSent >= Max of
        true ->
            {blocked, NewState#flow_state{send_blocked = true}};
        false ->
            {ok, NewState#flow_state{send_blocked = false}}
    end.

%% @doc Process a MAX_DATA frame from peer.
%% Updates our send limit.
-spec on_max_data_received(flow_state(), non_neg_integer()) -> flow_state().
on_max_data_received(#flow_state{send_max_data = OldMax} = State, NewMax) ->
    %% MAX_DATA only increases, never decreases
    ActualMax = max(OldMax, NewMax),
    State#flow_state{
        send_max_data = ActualMax,
        send_blocked = false
    }.

%% @doc Check if we're currently blocked on send flow control.
-spec send_blocked(flow_state()) -> boolean().
send_blocked(#flow_state{send_blocked = B}) -> B.

%%====================================================================
%% Receive Side
%%====================================================================

%% @doc Record that we received data.
-spec on_data_received(flow_state(), non_neg_integer()) ->
    {ok, flow_state()} | {error, flow_control_error}.
on_data_received(
    #flow_state{bytes_received = Received, recv_max_data = Max} = State,
    Size
) ->
    NewReceived = Received + Size,
    case NewReceived > Max of
        true ->
            {error, flow_control_error};
        false ->
            {ok, State#flow_state{bytes_received = NewReceived}}
    end.

%% @doc Check if we should send a MAX_DATA update.
%% Returns true if we've consumed more than the threshold.
-spec should_send_max_data(flow_state()) -> boolean().
should_send_max_data(#flow_state{
    bytes_received = Received,
    recv_max_data_sent = SentMax,
    initial_max_data = Initial
}) ->
    %% Send update when we've consumed > threshold of the window
    WindowConsumed = Received,
    WindowGranted = SentMax,
    Threshold = trunc(Initial * ?WINDOW_UPDATE_THRESHOLD),
    (WindowConsumed > WindowGranted - Threshold).

%% @doc Generate a new MAX_DATA value to send.
%% Returns {NewMaxData, UpdatedState}.
-spec generate_max_data(flow_state()) ->
    {non_neg_integer(), flow_state()}.
generate_max_data(
    #flow_state{
        bytes_received = Received,
        initial_max_data = Initial
    } = State
) ->
    %% Grant a new window based on bytes consumed
    NewMax = Received + Initial,
    NewState = State#flow_state{
        recv_max_data = NewMax,
        recv_max_data_sent = NewMax
    },
    {NewMax, NewState}.

%%====================================================================
%% Queries
%%====================================================================

%% @doc Get total bytes we've sent.
-spec bytes_sent(flow_state()) -> non_neg_integer().
bytes_sent(#flow_state{bytes_sent = B}) -> B.

%% @doc Get total bytes we've received.
-spec bytes_received(flow_state()) -> non_neg_integer().
bytes_received(#flow_state{bytes_received = B}) -> B.

%% @doc Get our current send limit (peer's MAX_DATA).
-spec send_limit(flow_state()) -> non_neg_integer().
send_limit(#flow_state{send_max_data = M}) -> M.

%% @doc Get our current receive limit (our MAX_DATA).
-spec recv_limit(flow_state()) -> non_neg_integer().
recv_limit(#flow_state{recv_max_data = M}) -> M.

%% @doc Get available send window.
-spec send_window(flow_state()) -> non_neg_integer().
send_window(#flow_state{bytes_sent = Sent, send_max_data = Max}) ->
    max(0, Max - Sent).

%% @doc Get available receive window.
-spec recv_window(flow_state()) -> non_neg_integer().
recv_window(#flow_state{bytes_received = Received, recv_max_data = Max}) ->
    max(0, Max - Received).