All checks were successful
Desktop source guards / guards (push) Successful in 8s
Every MTProto session through a WEB proxy is a stream on a single
carrier whose relay keeps one FIFO per direction. Bulk media and
uploads filled that FIFO, every session's own 4-8s receive timer then
fired at once, and the reconnects piled new OPENs behind the same
backlog.
- Uplink: a scheduler serves interactive streams first, download
requests next and uploads round-robin in 64 KiB frames, with at most
1 MiB of upload bytes uncredited by the relay (one bounded batch in
front of a chat request); a burst guard keeps uploads from starving.
- Downlink: download streams keep only a share of a 2 MiB credit budget
(256 KiB..1 MiB each) instead of the full 4 MiB window, so media waits
in the backend's TCP buffers rather than in front of chat replies.
- Liveness: a WEB session asks the carrier before dropping its stream.
It waits while its request is still queued or the downlink is busy,
fails when the pipe is quiet for 8s after delivery, busy for 30s, or
after 64s total. A carrier with bytes outstanding and no relay
progress for 20s is recovered once, for all streams.
- A WEB session opens one stream instead of racing identical ones, and
a -404 on a WEB stream reconnects first; only a second -404 before any
reply decrypts destroys the temporary key.
- Uploads and downloads use at most two sessions per DC through a WEB
proxy, and slow upload parts are no longer cancelled and moved.
- web_carrier diagnostics report carrier state, stalls, per-class
stream and queue counts, credit and throughput every 10s of activity.
The policy lives in web_proxy_flow.{h,cpp} and is covered by the new
test_web_proxy_flow target and a source guard.
367 lines
12 KiB
C++
367 lines
12 KiB
C++
/*
|
|
This file is part of Telegram Desktop,
|
|
the official desktop application for the Telegram messaging service.
|
|
|
|
For license and copyright information please follow this link:
|
|
https://github.com/telegramdesktop/tdesktop/blob/master/LEGAL
|
|
*/
|
|
#include "mtproto/proxy/control_plane.h"
|
|
|
|
#include "base/invoke_queued.h"
|
|
#include "mtproto/proxy/diagnostics.h"
|
|
#include "mtproto/proxy/transport_policy.h"
|
|
#include "mtproto/runtime/connection_status.h"
|
|
#include "mtproto/runtime/runtime_environment.h"
|
|
|
|
namespace MTP {
|
|
namespace {
|
|
constexpr auto kFreshRelaySuccessWindow = crl::time(15 * 1000);
|
|
|
|
[[nodiscard]] bool IsSuccess(const ProxyConnectionStatus &status) {
|
|
return status.phase == ProxyConnectionPhase::Connected;
|
|
}
|
|
|
|
[[nodiscard]] bool IsTerminalFailure(
|
|
const ProxyConnectionStatus &status) {
|
|
return (status.error != ProxyConnectionError::None)
|
|
|| IsMtproxyTerminalFailure(status.mtproxyReason);
|
|
}
|
|
|
|
[[nodiscard]] bool IsNewerProxyEpoch(
|
|
const ProxyConnectionAttempt ¤t,
|
|
const ProxyConnectionAttempt &update) {
|
|
if (current.runtimeId && update.runtimeId != current.runtimeId) {
|
|
return false;
|
|
}
|
|
if (update.proxyGeneration != current.proxyGeneration) {
|
|
return update.proxyGeneration > current.proxyGeneration;
|
|
}
|
|
if (update.proxyEpoch != current.proxyEpoch) {
|
|
return update.proxyEpoch > current.proxyEpoch;
|
|
}
|
|
return update.successEpoch > current.successEpoch;
|
|
}
|
|
|
|
[[nodiscard]] bool IsNewerAttempt(
|
|
const ProxyConnectionAttempt ¤t,
|
|
const ProxyConnectionAttempt &update) {
|
|
if (current.runtimeId && update.runtimeId != current.runtimeId) {
|
|
return false;
|
|
}
|
|
if (update.proxyGeneration != current.proxyGeneration) {
|
|
return update.proxyGeneration > current.proxyGeneration;
|
|
}
|
|
if (update.proxyEpoch != current.proxyEpoch) {
|
|
return update.proxyEpoch > current.proxyEpoch;
|
|
}
|
|
if (update.successEpoch != current.successEpoch) {
|
|
return update.successEpoch > current.successEpoch;
|
|
}
|
|
return update.attemptId > current.attemptId;
|
|
}
|
|
|
|
[[nodiscard]] bool IsOlderAttempt(
|
|
const ProxyConnectionAttempt ¤t,
|
|
const ProxyConnectionAttempt &update) {
|
|
if (current.runtimeId && update.runtimeId != current.runtimeId) {
|
|
return false;
|
|
}
|
|
if (current.proxyGeneration
|
|
&& update.proxyGeneration != current.proxyGeneration) {
|
|
return false;
|
|
}
|
|
if (current.proxyEpoch && !update.proxyEpoch) {
|
|
return true;
|
|
}
|
|
if (update.proxyEpoch && update.proxyEpoch < current.proxyEpoch) {
|
|
return true;
|
|
}
|
|
if (update.proxyEpoch != current.proxyEpoch) {
|
|
return false;
|
|
}
|
|
if (current.successEpoch && !update.successEpoch) {
|
|
return true;
|
|
}
|
|
if (update.successEpoch && update.successEpoch < current.successEpoch) {
|
|
return true;
|
|
}
|
|
if (update.successEpoch != current.successEpoch) {
|
|
return false;
|
|
}
|
|
return current.attemptId
|
|
&& update.attemptId
|
|
&& (update.attemptId < current.attemptId);
|
|
}
|
|
|
|
[[nodiscard]] bool IsOlderProxyGeneration(
|
|
const ProxyConnectionAttempt ¤t,
|
|
const ProxyConnectionAttempt &update) {
|
|
if (current.runtimeId && update.runtimeId != current.runtimeId) {
|
|
return false;
|
|
}
|
|
return current.proxyGeneration
|
|
&& (!update.proxyGeneration
|
|
|| (update.proxyGeneration < current.proxyGeneration));
|
|
}
|
|
|
|
[[nodiscard]] bool StickyWindowActive(
|
|
const ProxyConnectionStatus &status) {
|
|
return status.terminalUntil
|
|
&& (status.terminalUntil > crl::now());
|
|
}
|
|
|
|
[[nodiscard]] bool RelaySuccessIsFresh(
|
|
const ProxyConnectionStatus &status) {
|
|
return (status.phase == ProxyConnectionPhase::Connected)
|
|
&& status.successUntil
|
|
&& (status.successUntil > crl::now());
|
|
}
|
|
|
|
[[nodiscard]] bool IsRelayDataStall(
|
|
const ProxyConnectionStatus &status) {
|
|
switch (status.mtproxyReason) {
|
|
case ProxyMtproxyTerminalReason::ServerHelloOkNoAppData:
|
|
case ProxyMtproxyTerminalReason::ServerHelloOkNoMtprotoData:
|
|
case ProxyMtproxyTerminalReason::ConnectedNoMtprotoData:
|
|
case ProxyMtproxyTerminalReason::MtpReceiveTimeoutAfterData:
|
|
return true;
|
|
case ProxyMtproxyTerminalReason::None:
|
|
case ProxyMtproxyTerminalReason::DnsFailed:
|
|
case ProxyMtproxyTerminalReason::TcpConnectTimeout:
|
|
case ProxyMtproxyTerminalReason::TcpConnectedNoClientHelloWrite:
|
|
case ProxyMtproxyTerminalReason::ClientHelloSentNoServerHello:
|
|
case ProxyMtproxyTerminalReason::TlsAlertAfterClientHello:
|
|
case ProxyMtproxyTerminalReason::ServerHelloHmacMismatch:
|
|
case ProxyMtproxyTerminalReason::ServerHelloForeignTls:
|
|
case ProxyMtproxyTerminalReason::AppDataRemoteClosed:
|
|
case ProxyMtproxyTerminalReason::ProxyProtocolBadResponse:
|
|
return false;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
[[nodiscard]] bool ShadowedByFreshRelaySuccess(
|
|
const ProxyConnectionStatus ¤t,
|
|
const ProxyFact &fact) {
|
|
return RelaySuccessIsFresh(current)
|
|
&& IsTerminalFailure(fact.status)
|
|
&& !(fact.status.attempt == current.attempt)
|
|
&& !IsNewerProxyEpoch(current.attempt, fact.status.attempt);
|
|
}
|
|
|
|
[[nodiscard]] bool EmptyFact(const ProxyFact &fact) {
|
|
return (fact.status.phase == ProxyConnectionPhase::None)
|
|
&& !IsTerminalFailure(fact.status)
|
|
&& (fact.successScope == ProxyControlPlaneSuccessScope::None);
|
|
}
|
|
|
|
[[nodiscard]] ProxyDiagnosticsSource SourceForProxy(const ProxyData &proxy) {
|
|
return (proxy.type == ProxyData::Type::Mtproto)
|
|
? ProxyDiagnosticsSource::MTProxy
|
|
: ProxyDiagnosticsSource::Network;
|
|
}
|
|
|
|
void LogShadowedFact(
|
|
not_null<RuntimeEnvironment*> runtime,
|
|
const ProxyFact &fact) {
|
|
WriteProxyDiagnosticsLine(runtime, {
|
|
.source = SourceForProxy(fact.status.proxy),
|
|
.phase = ProxyDiagnosticsPhase::Failed,
|
|
.severity = ProxyDiagnosticsSeverity::Info,
|
|
.error = fact.status.error,
|
|
.mtproxyReason = fact.status.mtproxyReason,
|
|
.attempt = fact.status.attempt,
|
|
.terminalUntil = fact.status.terminalUntil,
|
|
.proxy = fact.status.proxy,
|
|
.message = u"proxy fact shadowed_by_fresh_success"_q,
|
|
});
|
|
}
|
|
|
|
[[nodiscard]] ProxyConnectionStatus ApplySelectedStatusUpdate(
|
|
const ProxyConnectionStatus ¤t,
|
|
ProxyConnectionStatus update) {
|
|
if (IsOlderProxyGeneration(current.attempt, update.attempt)) {
|
|
return current;
|
|
}
|
|
if (IsOlderAttempt(current.attempt, update.attempt)) {
|
|
return current;
|
|
}
|
|
if (RelaySuccessIsFresh(current)
|
|
&& IsTerminalFailure(update)
|
|
&& !(update.attempt == current.attempt)
|
|
&& !IsNewerProxyEpoch(current.attempt, update.attempt)) {
|
|
return current;
|
|
}
|
|
if (!IsMtproxyTerminalFailure(current.mtproxyReason)) {
|
|
return update;
|
|
}
|
|
if (IsSuccess(update)
|
|
|| IsMtproxyTerminalFailure(update.mtproxyReason)
|
|
|| IsNewerAttempt(current.attempt, update.attempt)) {
|
|
return update;
|
|
}
|
|
if (StickyWindowActive(current)) {
|
|
return current;
|
|
}
|
|
return update;
|
|
}
|
|
|
|
} // namespace
|
|
|
|
ProxyControlPlane::ProxyControlPlane(not_null<RuntimeEnvironment*> runtime)
|
|
: _runtime(runtime) {
|
|
}
|
|
|
|
ProxyConnectionStatus ProxyControlPlane::selectedStatus() const {
|
|
return _selectedStatus;
|
|
}
|
|
|
|
ProxyEndpointSnapshot ProxyControlPlane::endpointSnapshot() const {
|
|
return _endpointSnapshot;
|
|
}
|
|
|
|
ProxyFact ProxyControlPlane::FactFromReport(
|
|
const ProxyEventReport &report) {
|
|
auto fact = ProxyFact();
|
|
fact.status = {
|
|
.phase = ProxyConnectionPhase::None,
|
|
.error = report.error,
|
|
.mtproxyReason = report.mtproxyReason,
|
|
.attempt = report.attempt,
|
|
.terminalUntil = report.terminalUntil,
|
|
.proxy = report.proxy,
|
|
};
|
|
switch (report.phase) {
|
|
case ProxyDiagnosticsPhase::Resolving:
|
|
fact.status.phase = ProxyConnectionPhase::Resolving;
|
|
return fact;
|
|
case ProxyDiagnosticsPhase::Connecting:
|
|
case ProxyDiagnosticsPhase::TcpConnected:
|
|
fact.status.phase = ProxyConnectionPhase::Connecting;
|
|
return fact;
|
|
case ProxyDiagnosticsPhase::ClientHelloSent:
|
|
fact.status.phase = ProxyConnectionPhase::Handshake;
|
|
return fact;
|
|
case ProxyDiagnosticsPhase::ServerHelloOk:
|
|
fact.status.phase = ProxyConnectionPhase::CheckingTelegram;
|
|
fact.successScope = ProxyControlPlaneSuccessScope::Handshake;
|
|
return fact;
|
|
case ProxyDiagnosticsPhase::TelegramCheck:
|
|
fact.status.phase = ProxyConnectionPhase::CheckingTelegram;
|
|
return fact;
|
|
case ProxyDiagnosticsPhase::Connected:
|
|
fact.status.phase = ProxyConnectionPhase::CheckingTelegram;
|
|
fact.successScope = ProxyControlPlaneSuccessScope::Handshake;
|
|
return fact;
|
|
case ProxyDiagnosticsPhase::MtpFirstDataReceived:
|
|
fact.status.phase = ProxyConnectionPhase::Connected;
|
|
fact.successScope = ProxyControlPlaneSuccessScope::Relay;
|
|
return fact;
|
|
case ProxyDiagnosticsPhase::Failed:
|
|
fact.status.phase = ProxyConnectionPhase::Failed;
|
|
return fact;
|
|
case ProxyDiagnosticsPhase::ProxyCheckStarted:
|
|
case ProxyDiagnosticsPhase::ProxyCheckFinished:
|
|
fact.status.error = ProxyConnectionError::None;
|
|
fact.status.mtproxyReason = ProxyMtproxyTerminalReason::None;
|
|
return fact;
|
|
case ProxyDiagnosticsPhase::None:
|
|
case ProxyDiagnosticsPhase::AdmissionQueued:
|
|
case ProxyDiagnosticsPhase::AdmissionStarted:
|
|
case ProxyDiagnosticsPhase::AdmissionCancelled:
|
|
case ProxyDiagnosticsPhase::RouteSelected:
|
|
case ProxyDiagnosticsPhase::RouteFailed:
|
|
case ProxyDiagnosticsPhase::CanonicalDegraded:
|
|
case ProxyDiagnosticsPhase::CanonicalRecovered:
|
|
case ProxyDiagnosticsPhase::StealthRecipeApplied:
|
|
case ProxyDiagnosticsPhase::TransportFallbackApplied:
|
|
case ProxyDiagnosticsPhase::RotationSwitched:
|
|
case ProxyDiagnosticsPhase::MtpConnecting:
|
|
case ProxyDiagnosticsPhase::MtpTransportReady:
|
|
case ProxyDiagnosticsPhase::MtpKeyCreating:
|
|
case ProxyDiagnosticsPhase::MtpKeyReady:
|
|
case ProxyDiagnosticsPhase::MtpReceiveTimeout:
|
|
case ProxyDiagnosticsPhase::MtpConnectTimeout:
|
|
case ProxyDiagnosticsPhase::MtpBrokerTimeout:
|
|
case ProxyDiagnosticsPhase::MtpPingTimeout:
|
|
case ProxyDiagnosticsPhase::MtpBindFailed:
|
|
case ProxyDiagnosticsPhase::MtpKeyDestroyed:
|
|
case ProxyDiagnosticsPhase::MtpRestart:
|
|
case ProxyDiagnosticsPhase::AttemptSummary:
|
|
case ProxyDiagnosticsPhase::Liveness:
|
|
case ProxyDiagnosticsPhase::WebCarrier:
|
|
fact.status.error = ProxyConnectionError::None;
|
|
fact.status.mtproxyReason = ProxyMtproxyTerminalReason::None;
|
|
return fact;
|
|
}
|
|
return fact;
|
|
}
|
|
|
|
ProxyConnectionStatus ProxyControlPlane::Reduce(
|
|
const ProxyConnectionStatus ¤t,
|
|
ProxyFact fact) {
|
|
if (EmptyFact(fact)) {
|
|
return current;
|
|
}
|
|
if (IsProxyCheck(fact.status.attempt.use)) {
|
|
return current;
|
|
}
|
|
if (IsRelayDataStall(fact.status)) {
|
|
fact.status.error = ProxyConnectionError::None;
|
|
}
|
|
if (fact.status.proxy.type == ProxyData::Type::Mtproto) {
|
|
if (fact.successScope == ProxyControlPlaneSuccessScope::Relay) {
|
|
fact.status.phase = ProxyConnectionPhase::Connected;
|
|
fact.status.successUntil = crl::now() + kFreshRelaySuccessWindow;
|
|
} else if (ShadowedByFreshRelaySuccess(current, fact)) {
|
|
// One mtproxy carries every session of every account, so a dozen
|
|
// sockets are always in flight against it and one of them is
|
|
// almost always retrying. Reporting that socket's failure while
|
|
// the others relay Telegram data describes the socket, not the
|
|
// proxy, and it is the proxy the shield claims to be about.
|
|
return current;
|
|
}
|
|
return std::move(fact.status);
|
|
}
|
|
if (fact.successScope == ProxyControlPlaneSuccessScope::Relay) {
|
|
fact.status.phase = ProxyConnectionPhase::Connected;
|
|
fact.status.successUntil = crl::now()
|
|
+ kFreshRelaySuccessWindow;
|
|
}
|
|
if (ShadowedByFreshRelaySuccess(current, fact)) {
|
|
return current;
|
|
}
|
|
return ApplySelectedStatusUpdate(
|
|
current,
|
|
std::move(fact.status));
|
|
}
|
|
|
|
void ProxyControlPlane::submitFact(const ProxyEventReport &report) {
|
|
auto fact = FactFromReport(report);
|
|
if (EmptyFact(fact)) {
|
|
return;
|
|
}
|
|
InvokeQueued(_runtime, [=] { submitFactOnOwner(fact); });
|
|
}
|
|
|
|
void ProxyControlPlane::submitFactOnOwner(ProxyFact fact) {
|
|
// Every proxy type, mtproxy included, now reduces into the same
|
|
// connection status. There is no separate per-endpoint projection to
|
|
// divert into: what the user sees is what the transport reported.
|
|
const auto current = _runtime->instance().connectionStatus
|
|
? _runtime->instance().connectionStatus->proxyStatus()
|
|
: ProxyConnectionStatus();
|
|
if (ShadowedByFreshRelaySuccess(current, fact)) {
|
|
LogShadowedFact(_runtime, fact);
|
|
}
|
|
if (_runtime->instance().connectionStatus) {
|
|
const auto reduced = ProxyControlPlane::Reduce(current, fact);
|
|
_selectedStatus = reduced;
|
|
_endpointSnapshot.proxy = reduced.proxy;
|
|
_endpointSnapshot.status = reduced;
|
|
_endpointSnapshot.relayProven = RelaySuccessIsFresh(reduced);
|
|
_runtime->instance().connectionStatus->setProxyStatus(reduced);
|
|
}
|
|
}
|
|
|
|
} // namespace MTP
|