diff --git a/doc/handler-downstream-flow.md b/doc/handler-downstream-flow.md
new file mode 100644
index 0000000..9a48dac
--- /dev/null
+++ b/doc/handler-downstream-flow.md
@@ -0,0 +1,45 @@
+# Handler ↔ downstream lookup and handshake
+
+Shows how `mtp_handler` locates an `mtp_down_conn` for a new client connection
+and the steady-state data flow that follows.
+
+**Key actors:**
+- `mtp_handler` — one process per Telegram client TCP connection
+- `mtp_dc_pool` — manages a pool of downstream connections for one DC
+- `mtp_down_conn` — multiplexed TCP connection to a Telegram DC
+- `Telegram DC` — the upstream Telegram data-centre server
+
+```mermaid
+sequenceDiagram
+ participant Client as Telegram client
+ participant Handler as mtp_handler
+ participant Pool as mtp_dc_pool
+ participant Down as mtp_down_conn
+ participant TG as Telegram DC
+
+ Client->>Handler: TCP connect + Hello bytes
+
+ Note over Handler: decode protocol headers
(fake-TLS / obfuscated / secure)
stage: hello → tunnel
+
+ Note over Handler: resolve pool: whereis(dc_to_pool_name(DcId))
(registered name lookup; falls back to default DC from mtp_config if not found)
+ Handler->>Pool: mtp_dc_pool:get(Pool, self(), Opts) [sync]
+ Pool-->>Down: upstream_new(Handler, Opts) [cast]
+ Pool->>Handler: Downstream pid
+
+ Note over Handler: down = Downstream
stage = tunnel
+
+ loop steady-state data exchange
+ Client->>Handler: TCP data
+ Handler->>Down: mtp_down_conn:send(Down, Data) [sync]
+ Down->>TG: TCP data (RPC-framed)
+ TG->>Down: TCP data
+ Down->>Handler: ok
+ Down-->>Handler: {proxy_ans, Down, Data} [cast]
+ Handler->>Client: TCP data
+ Handler-->>Down: mtp_down_conn:ack(Down, Count, Size) [cast]
+ end
+
+ Client->>Handler: TCP close
+ Handler-->>Pool: mtp_dc_pool:return(Pool, self()) [cast]
+ Pool-->>Down: upstream_closed(Down, Handler) [cast]
+```
diff --git a/src/mtp_config.erl b/src/mtp_config.erl
index b2420fb..f1aeb00 100644
--- a/src/mtp_config.erl
+++ b/src/mtp_config.erl
@@ -18,8 +18,8 @@
-export([get_downstream_safe/2,
get_downstream_pool/1,
get_netloc/1,
- get_netloc_safe/1,
get_secret/0,
+ get_default_dc/0,
status/0,
update/0]).
@@ -37,6 +37,7 @@
-define(TAB, ?MODULE).
-define(IPS_KEY(DcId), {id, DcId}).
-define(IDS_KEY, dc_ids).
+-define(DEFAULT_DC_KEY, default_dc).
-define(SECRET_URL, "https://core.telegram.org/getProxySecret").
-define(CONFIG_URL, "https://core.telegram.org/getProxyConfig").
@@ -65,9 +66,11 @@ get_downstream_safe(DcId, Opts) ->
error({pool_empty, DcId, Pool})
end;
not_found ->
- [{?IDS_KEY, L}] = ets:lookup(?TAB, ?IDS_KEY),
- NewDcId = random_choice(L),
- get_downstream_safe(NewDcId, Opts)
+ case get_default_dc() of
+ undefined -> error({no_pool, DcId});
+ DcId -> error({no_pool, DcId}); % default == requested, avoid loop
+ NewDcId -> get_downstream_safe(NewDcId, Opts)
+ end
end.
get_downstream_pool(DcId) ->
@@ -78,17 +81,6 @@ get_downstream_pool(DcId) ->
not_found
end.
--spec get_netloc_safe(dc_id()) -> {dc_id(), netloc()}.
-get_netloc_safe(DcId) ->
- case get_netloc(DcId) of
- {ok, Addr} -> {DcId, Addr};
- not_found ->
- [{?IDS_KEY, L}] = ets:lookup(?TAB, ?IDS_KEY),
- NewDcId = random_choice(L),
- %% Get random DC; it might return 0 and recurse aggain
- get_netloc_safe(NewDcId)
- end.
-
get_netloc(DcId) ->
Key = ?IPS_KEY(DcId),
case ets:lookup(?TAB, Key) of
@@ -107,6 +99,13 @@ get_secret() ->
[{_, Key}] = ets:lookup(?TAB, key),
Key.
+-spec get_default_dc() -> dc_id() | undefined.
+get_default_dc() ->
+ case ets:lookup(?TAB, ?DEFAULT_DC_KEY) of
+ [{?DEFAULT_DC_KEY, DcId}] -> DcId;
+ [] -> undefined
+ end.
+
-spec status() -> [mtp_dc_pool:status()].
status() ->
[{?IDS_KEY, L}] = ets:lookup(?TAB, ?IDS_KEY),
@@ -130,7 +129,7 @@ init([]) ->
#{timeout => {env, ?APP, conf_refresh_interval, 3600},
unit => second}),
Tab = ets:new(?TAB, [set,
- public,
+ protected,
named_table,
{read_concurrency, true}]),
State = #state{tab = Tab,
@@ -189,17 +188,23 @@ update_key(Tab) ->
update_config(Tab) ->
Url = application:get_env(mtproto_proxy, proxy_config_url, ?CONFIG_URL),
{ok, Body} = http_get(Url),
- Downstreams = parse_config(Body),
+ {DefaultDc, Downstreams} = parse_config(Body),
update_downstreams(Downstreams, Tab),
- update_ids(Downstreams, Tab).
+ update_ids(Downstreams, DefaultDc, Tab).
parse_config(Body) ->
Lines = string:lexemes(Body, "\n"),
- ProxyLines = lists:filter(
- fun("proxy_for " ++ _) -> true;
- (_) -> false
- end, Lines),
- [parse_downstream(Line) || Line <- ProxyLines].
+ {DefaultDc, Downstreams} =
+ lists:foldl(
+ fun("default " ++ _ = Line, {_, Ds}) ->
+ ["default", DcIdStr] = string:lexemes(Line, " ;"),
+ {list_to_integer(DcIdStr), Ds};
+ ("proxy_for " ++ _ = Line, {Def, Ds}) ->
+ {Def, [parse_downstream(Line) | Ds]};
+ (_, Acc) ->
+ Acc
+ end, {undefined, []}, Lines),
+ {DefaultDc, lists:reverse(Downstreams)}.
parse_downstream(Line) ->
["proxy_for",
@@ -231,9 +236,10 @@ update_downstreams(Downstreams, Tab) ->
end,
maps:keys(ByDc)).
-update_ids(Downstreams, Tab) ->
+update_ids(Downstreams, DefaultDc, Tab) ->
Ids = lists:usort([DcId || {DcId, _, _} <- Downstreams]),
- true = ets:insert(Tab, {?IDS_KEY, Ids}).
+ true = ets:insert(Tab, {?IDS_KEY, Ids}),
+ true = ets:insert(Tab, {?DEFAULT_DC_KEY, DefaultDc}).
update_ip() ->
case application:get_env(?APP, ip_lookup_services) of
@@ -283,15 +289,22 @@ random_choice(L) ->
-include_lib("eunit/include/eunit.hrl").
parse_test() ->
- Config = ("# force_probability 1 10
-proxy_for 1 149.154.175.50:8888;
-proxy_for -1 149.154.175.50:8888;
-proxy_for 2 149.154.162.39:80;
-proxy_for 2 149.154.162.33:80;"),
+ Config = ("# force_probability 1 10\n"
+ "default 2;\n"
+ "proxy_for 1 149.154.175.50:8888;\n"
+ "proxy_for -1 149.154.175.50:8888;\n"
+ "proxy_for 2 149.154.162.39:80;\n"
+ "proxy_for 2 149.154.162.33:80;"),
Expect = [{1, {149, 154, 175, 50}, 8888},
{-1, {149, 154, 175, 50}, 8888},
{2, {149, 154, 162, 39}, 80},
{2, {149, 154, 162, 33},80}],
- ?assertEqual(Expect, parse_config(Config)).
+ ?assertEqual({2, Expect}, parse_config(Config)).
+
+parse_no_default_test() ->
+ Config = ("proxy_for 1 149.154.175.50:8888;\n"
+ "proxy_for 2 149.154.162.39:80;"),
+ {DefaultDc, _} = parse_config(Config),
+ ?assertEqual(undefined, DefaultDc).
-endif.