From ed2fa40ee58c3b852ef51e5309ecde2c07cc2310 Mon Sep 17 00:00:00 2001 From: Sergey Prokhorov Date: Tue, 7 Apr 2026 19:12:59 +0200 Subject: [PATCH] Add TCP keepalive tuning and 4-element UpsStatic - TCP keepalive on downstream connections (KEEPIDLE=40s, KEEPINTVL=40s, KEEPCNT=5): matches reference C implementation (net-events.c). Uses raw socket options for OTP 25+ compatibility (named keepidle/ keepintvl/keepcnt require OTP 28.3+/Kernel 10.5). - Expand UpsStatic to 4-element tuple {ConnId, Addr, AdTag, PacketLayer} and thread packet_layer through mtp_handler -> mtp_dc_pool -> mtp_down_conn. Fixes pattern match in handle_upstream_closed. - Reduce log noise for expected DC connection rotation: When Telegram closes a downstream connection with no active clients, stop with {shutdown, downstream_socket_closed} instead of bare atom; mtp_down_conn logs INFO, mtp_dc_pool logs INFO (was ERROR for all). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/mtp_dc_pool.erl | 10 ++++++-- src/mtp_down_conn.erl | 54 +++++++++++++++++++++++++++++++++---------- src/mtp_handler.erl | 3 ++- 3 files changed, 52 insertions(+), 15 deletions(-) diff --git a/src/mtp_dc_pool.erl b/src/mtp_dc_pool.erl index fed56a8..8e95682 100644 --- a/src/mtp_dc_pool.erl +++ b/src/mtp_dc_pool.erl @@ -72,7 +72,8 @@ dc_to_pool_name(DcId) -> binary_to_atom(<<"mtp_dc_pool_", (integer_to_binary(DcId))/binary>>, utf8). -spec get(pid(), upstream(), #{addr := mtp_config:netloc_v4v6(), - ad_tag => binary()}) -> downstream() | {error, atom()}. + ad_tag => binary(), + packet_layer => mtp_down_conn:packet_layer()}) -> downstream() | {error, atom()}. get(Pool, Upstream, #{addr := _} = Opts) -> gen_server:call(Pool, {get, Upstream, Opts}). @@ -192,7 +193,12 @@ handle_down(MonRef, Pid, Reason, #state{downstreams = Ds, {Pid, DsM1} -> Pending1 = lists:delete(Pid, Pending), Ds1 = ds_remove(Pid, Ds), - ?LOG_ERROR("Downstream=~p is down. reason=~p", [Pid, Reason]), + case Reason of + {shutdown, downstream_socket_closed} -> + ?LOG_INFO("Downstream=~p closed (no active clients)", [Pid]); + _ -> + ?LOG_ERROR("Downstream=~p is down. reason=~p", [Pid, Reason]) + end, maybe_restart_connection( St#state{pending_downstreams = Pending1, downstreams = Ds1, diff --git a/src/mtp_down_conn.erl b/src/mtp_down_conn.erl index 06bcf61..77534a7 100644 --- a/src/mtp_down_conn.erl +++ b/src/mtp_down_conn.erl @@ -25,7 +25,7 @@ %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). --export_type([handle/0, upstream_opts/0]). +-export_type([handle/0, upstream_opts/0, packet_layer/0]). -include_lib("kernel/include/logger.hrl"). @@ -38,13 +38,24 @@ -define(MAX_CODEC_BUFFERS, 5 * 1024 * 1024). -define(DEFAULT_CLIENTS_PER_CONN, 300). +%% Linux raw socket option constants for TCP keepalive tuning. +%% Named options (keepidle/keepintvl/keepcnt) require OTP 28.3+ (Kernel 10.5); +%% raw options work on OTP 25+ on any Linux kernel. +-define(IPPROTO_TCP, 6). +-define(TCP_KEEPIDLE, 4). +-define(TCP_KEEPINTVL, 5). +-define(TCP_KEEPCNT, 6). + -type handle() :: pid(). +-type packet_layer() :: mtp_abridged | mtp_intermediate | mtp_secure. -type upstream_opts() :: #{addr := mtp_config:netloc_v4v6(), % IP/Port of TG client - ad_tag => binary()}. + ad_tag => binary(), + packet_layer => packet_layer()}. -type upstream() :: { _UpsStatic ::{_ConnId :: mtp_rpc:conn_id(), _Addr :: binary(), - _AdTag :: binary() | undefined}, + _AdTag :: binary() | undefined, + _Protocol :: mtp_abridged | mtp_intermediate | mtp_secure}, _NonAckCount :: non_neg_integer(), _NonAckBytes :: non_neg_integer() }. @@ -155,8 +166,11 @@ handle_info({tcp, Sock, Data}, #state{sock = Sock, dc_id = DcId} = S) -> {ok, S1} = handle_downstream_data(Data, S), activate_if_no_overflow(S1), {noreply, S1}; -handle_info({tcp_closed, Sock}, #state{sock = Sock} = State) -> - {stop, downstream_socket_closed, State}; +handle_info({tcp_closed, Sock}, #state{sock = Sock, upstreams = Ups} = State) -> + case map_size(Ups) of + 0 -> {stop, {shutdown, downstream_socket_closed}, State}; + _ -> {stop, downstream_socket_closed, State} + end; handle_info({tcp_error, Sock, Reason}, #state{sock = Sock} = State) -> {stop, {downstream_tcp_error, Reason}, State}; handle_info(do_connect, #state{dc_id = DcId} = State) -> @@ -180,10 +194,15 @@ handle_info(handshake_timeout, #state{stage = Stage, dc_id = DcId} = St) -> end. -terminate(_Reason, #state{upstreams = Ups}) -> - %% Should I do this or dc_pool? Maybe only when reason is 'normal'? - ?LOG_WARNING("Downstream terminates with reason ~p; len(upstreams)=~p", - [_Reason, map_size(Ups)]), +terminate(Reason, #state{upstreams = Ups}) -> + NUps = map_size(Ups), + case Reason of + {shutdown, downstream_socket_closed} -> + ?LOG_INFO("Downstream closed (no active clients); len(upstreams)=~p", [NUps]); + _ -> + ?LOG_WARNING("Downstream terminates with reason ~p; len(upstreams)=~p", + [Reason, NUps]) + end, Self = self(), lists:foreach( fun(Upstream) -> @@ -211,7 +230,8 @@ handle_upstream_new(Upstream, Opts, #state{upstreams = Ups, ConnId = erlang:unique_integer(), {Ip, Port} = maps:get(addr, Opts), AdTag = maps:get(ad_tag, Opts, undefined), - UpsStatic = {ConnId, iolist_to_binary(mtp_rpc:encode_ip_port(Ip, Port)), AdTag}, + PacketLayer = maps:get(packet_layer, Opts, mtp_abridged), + UpsStatic = {ConnId, iolist_to_binary(mtp_rpc:encode_ip_port(Ip, Port)), AdTag, PacketLayer}, Ups1 = Ups#{Upstream => {UpsStatic, 0, 0}}, UpsRev1 = UpsRev#{ConnId => Upstream}, ?LOG_DEBUG("New upstream=~p conn_id=~p", [Upstream, ConnId]), @@ -223,7 +243,7 @@ handle_upstream_closed(Upstream, #state{upstreams = Ups, upstreams_rev = UpsRev} = St) -> %% See "mtproto-proxy.c:remove_ext_connection case maps:take(Upstream, Ups) of - {{{ConnId, _, _}, _, _}, Ups1} -> + {{{ConnId, _, _, _}, _, _}, Ups1} -> St1 = non_ack_cleanup_upstream(Upstream, St), UpsRev1 = maps:remove(ConnId, UpsRev), St2 = St1#state{upstreams = Ups1, @@ -287,6 +307,9 @@ handle_rpc({close_ext, ConnId}, St) -> end; handle_rpc({simple_ack, ConnId, Confirm}, S) -> up_send({simple_ack, self(), Confirm}, ConnId, S); +handle_rpc({ping, PingId}, S) -> + {ok, S1} = down_send(mtp_rpc:encode_packet({pong, PingId}, undefined), S), + S1; handle_rpc({unknown, Tag, Tail}, S) -> ?LOG_INFO("Unknown packet from backend. Tag ~w, tail: ~w", [Tag, Tail]), S. @@ -472,13 +495,20 @@ connect(DcId, S) -> tcp_connect(Host, Port) -> BufSize = application:get_env(?APP, downstream_socket_buffer_size, ?MAX_SOCK_BUF_SIZE), + %% TCP keepalive tuning: match reference implementations (MTProxy C, mtprotoproxy Python). + %% Start probing after 40s idle, retry every 40s, drop after 5 failed probes. + %% Raw options are used for OTP 25+ compatibility (named keepidle/keepintvl/keepcnt + %% require OTP 28.3+ / Kernel 10.5). SockOpts = [{active, once}, {packet, raw}, {mode, binary}, {buffer, BufSize}, {send_timeout, ?SEND_TIMEOUT}, %% {nodelay, true}, - {keepalive, true}], + {keepalive, true}, + {raw, ?IPPROTO_TCP, ?TCP_KEEPIDLE, <<40:32/native>>}, + {raw, ?IPPROTO_TCP, ?TCP_KEEPINTVL, <<40:32/native>>}, + {raw, ?IPPROTO_TCP, ?TCP_KEEPCNT, <<5:32/native>>}], case gen_tcp:connect(Host, Port, SockOpts, ?CONN_TIMEOUT) of {ok, Sock} -> {ok, Sock}; diff --git a/src/mtp_handler.erl b/src/mtp_handler.erl index 77e4044..56df270 100644 --- a/src/mtp_handler.erl +++ b/src/mtp_handler.erl @@ -419,7 +419,8 @@ parse_upstream_data(<>, Codec2 = mtp_codec:replace(packet, PacketLayerMod, PacketCodec, Codec1), Codec = mtp_codec:push_back(crypto, Rest, Codec2), Opts = #{ad_tag => Tag, - addr => Addr}, + addr => Addr, + packet_layer => PacketLayerMod}, {RealDcId, Pool, Downstream} = mtp_config:get_downstream_safe(DcId, Opts), handle_upstream_data( <<>>,