Tonproxy improvements (ton-blockchain#483)

* Bugfixes in rldp-http-proxy and http parser

* Tonlib: change liteservers on query timeout or connection close

* Increase maximum size of http request

* Minor bugfixes in http

Author: SpyCheese

adnl/adnl-ext-client.hpp

@@ -80,6 +80,9 @@ class AdnlExtClientImpl : public AdnlExtClient {
     if (!conn_.empty() && conn_.get() == conn) {
       callback_->on_stop_ready();
       conn_ = {};
+      for (auto& q : out_queries_) {
+        td::actor::send_closure(q.second, &AdnlQuery::set_error, td::Status::Error(ErrorCode::cancelled));
+      }
       alarm_timestamp() = next_create_at_;
       try_stop();
     }

adnl/adnl-query.cpp

@@ -25,13 +25,16 @@ namespace ton {
 namespace adnl {
 
 void AdnlQuery::alarm() {
-  promise_.set_error(td::Status::Error(ErrorCode::timeout, "adnl query timeout"));
-  stop();
+  set_error(td::Status::Error(ErrorCode::timeout, "adnl query timeout"));
 }
 void AdnlQuery::result(td::BufferSlice data) {
   promise_.set_value(std::move(data));
   stop();
 }
+void AdnlQuery::set_error(td::Status error) {
+  promise_.set_error(std::move(error));
+  stop();
+}
 
 AdnlQueryId AdnlQuery::random_query_id() {
   AdnlQueryId q_id;

adnl/adnl-query.h

@@ -48,6 +48,7 @@ class AdnlQuery : public td::actor::Actor {
   }
   void alarm() override;
   void result(td::BufferSlice data);
+  void set_error(td::Status error);
   void start_up() override {
     alarm_timestamp() = timeout_;
   }

http/http-inbound-connection.cpp

@@ -79,10 +79,10 @@ td::Status HttpInboundConnection::receive(td::ChainBufferReader &input) {
       send_client_error();
       return td::Status::OK();
     }
+    cur_request_ = R.move_as_ok();
     if (exit_loop) {
       return td::Status::OK();
     }
-    cur_request_ = R.move_as_ok();
   }
 
   auto payload = cur_request_->create_empty_payload().move_as_ok();

http/http-outbound-connection.cpp

@@ -42,10 +42,10 @@ td::Status HttpOutboundConnection::receive(td::ChainBufferReader &input) {
       answer_error(HttpStatusCode::status_bad_request, "", std::move(promise_));
       return td::Status::OK();
     }
+    cur_response_ = R.move_as_ok();
     if (exit_loop) {
       return td::Status::OK();
     }
-    cur_response_ = R.move_as_ok();
   }
 
   if (cur_response_->code() == 100) {

http/http.cpp

@@ -279,25 +279,20 @@ td::Status HttpPayload::parse(td::ChainBufferReader &input) {
       } break;
       case ParseState::reading_chunk_data: {
         if (cur_chunk_size_ == 0) {
-          switch (type_) {
-            case PayloadType::pt_empty:
-              UNREACHABLE();
-            case PayloadType::pt_eof:
-            case PayloadType::pt_tunnel:
-              cur_chunk_size_ = 1LL << 60;
-              break;
-            case PayloadType::pt_chunked:
-              state_ = ParseState::reading_crlf;
-              break;
-            case PayloadType::pt_content_length: {
-              LOG(INFO) << "payload parse success";
-              const std::lock_guard<std::mutex> lock{mutex_};
-              state_ = ParseState::completed;
-              run_callbacks();
-              return td::Status::OK();
-            } break;
+          if (type_ == PayloadType::pt_eof || type_ == PayloadType::pt_tunnel) {
+            cur_chunk_size_ = 1LL << 60;
+          } else if (type_ == PayloadType::pt_chunked) {
+            state_ = ParseState::reading_crlf;
+            break;
+          } else if (type_ == PayloadType::pt_content_length) {
+            LOG(INFO) << "payload parse success";
+            const std::lock_guard<std::mutex> lock{mutex_};
+            state_ = ParseState::completed;
+            run_callbacks();
+            return td::Status::OK();
+          } else {
+            UNREACHABLE();
           }
-          break;
         }
         if (input.size() == 0) {
           return td::Status::OK();
@@ -502,7 +497,7 @@ bool HttpPayload::store_http(td::ChainBufferWriter &output, size_t max_size, Htt
       char buf[64];
       ::sprintf(buf, "%lx\r\n", s.size());
       auto slice = td::Slice(buf, strlen(buf));
-      wrote |= !slice.empty();
+      wrote = true;
       output.append(slice);
     }
 
@@ -514,7 +509,8 @@ bool HttpPayload::store_http(td::ChainBufferWriter &output, size_t max_size, Htt
       wrote = true;
     }
   }
-  if (chunks_.size() != 0 || !parse_completed()) {
+  auto cur_state = state_.load(std::memory_order_consume);
+  if (chunks_.size() != 0 || (cur_state != ParseState::reading_trailer && cur_state != ParseState::completed)) {
     return wrote;
   }
   if (!written_zero_chunk_) {
@@ -531,7 +527,7 @@ bool HttpPayload::store_http(td::ChainBufferWriter &output, size_t max_size, Htt
   }
 
   while (max_size > 0) {
-    auto cur_state = state_.load(std::memory_order_consume);
+    cur_state = state_.load(std::memory_order_consume);
     HttpHeader h = get_header();
     if (h.empty()) {
       if (cur_state != ParseState::completed) {
@@ -587,7 +583,8 @@ tl_object_ptr<ton_api::http_payloadPart> HttpPayload::store_tl(size_t max_size)
     max_size -= s.size();
   }
   obj->data_.truncate(obj->data_.size() - S.size());
-  if (chunks_.size() != 0) {
+  auto cur_state = state_.load(std::memory_order_consume);
+  if (chunks_.size() != 0 || (cur_state != ParseState::reading_trailer && cur_state != ParseState::completed)) {
     return obj;
   }
   if (!written_zero_chunk_) {
@@ -597,7 +594,7 @@ tl_object_ptr<ton_api::http_payloadPart> HttpPayload::store_tl(size_t max_size)
   LOG(INFO) << "data completed";
 
   while (max_size > 0) {
-    auto cur_state = state_.load(std::memory_order_consume);
+    cur_state = state_.load(std::memory_order_consume);
     HttpHeader h = get_header();
     if (h.empty()) {
       if (cur_state != ParseState::completed) {
@@ -869,7 +866,7 @@ td::Status HttpHeader::basic_check() {
   }
   for (auto &c : value) {
     if (c == '\r' || c == '\n') {
-      return td::Status::Error("bad character in header name");
+      return td::Status::Error("bad character in header value");
     }
   }
   return td::Status::OK();

rldp-http-proxy/DNSResolver.cpp

@@ -25,6 +25,7 @@
 */
 #include "DNSResolver.h"
 #include "td/utils/overloaded.h"
+#include "common/delay.h"
 
 static const double CACHE_TIMEOUT_HARD = 300.0;
 static const double CACHE_TIMEOUT_SOFT = 270.0;
@@ -33,8 +34,18 @@ DNSResolver::DNSResolver(td::actor::ActorId<TonlibClient> tonlib_client) : tonli
 }
 
 void DNSResolver::start_up() {
+  sync();
+}
+
+void DNSResolver::sync() {
   auto obj = tonlib_api::make_object<tonlib_api::sync>();
-  auto P = td::PromiseCreator::lambda([](td::Result<tonlib_api::object_ptr<tonlib_api::Object>>) {});
+  auto P = td::PromiseCreator::lambda([SelfId =
+                                           actor_id(this)](td::Result<tonlib_api::object_ptr<tonlib_api::Object>> R) {
+    if (R.is_error()) {
+      LOG(WARNING) << "Sync error: " << R.move_as_error();
+      ton::delay_action([SelfId]() { td::actor::send_closure(SelfId, &DNSResolver::sync); }, td::Timestamp::in(5.0));
+    }
+  });
   td::actor::send_closure(tonlib_client_, &TonlibClient::send_request, std::move(obj), std::move(P));
 }
 

rldp-http-proxy/DNSResolver.h

@@ -37,6 +37,7 @@ class DNSResolver : public td::actor::Actor {
   void resolve(std::string host, td::Promise<ton::adnl::AdnlNodeIdShort> promise);
 
  private:
+  void sync();
   void save_to_cache(std::string host, ton::adnl::AdnlNodeIdShort id);
 
   td::actor::ActorId<TonlibClient> tonlib_client_;

rldp-http-proxy/rldp-http-proxy.cpp

@@ -117,7 +117,7 @@ class HttpRemote : public td::actor::Actor {
             }
           });
       td::actor::send_closure(client_, &ton::http::HttpClient::send_request, std::move(request), std::move(payload),
-                              td::Timestamp::in(30.0), std::move(P));
+                              td::Timestamp::never(), std::move(P));
     } else {
       ton::http::answer_error(ton::http::HttpStatusCode::status_bad_request, "", std::move(promise));
     }
@@ -801,6 +801,7 @@ class RldpToTcpRequestSender : public td::actor::Actor {
       , dst_(dst)
       , request_(std::move(request))
       , request_payload_(std::move(request_payload))
+      , proto_version_(request_->proto_version())
       , promise_(std::move(promise))
       , adnl_(adnl)
       , rldp_(rldp)
@@ -824,19 +825,17 @@ class RldpToTcpRequestSender : public td::actor::Actor {
   }
 
   void got_result(std::pair<std::unique_ptr<ton::http::HttpResponse>, std::shared_ptr<ton::http::HttpPayload>> R) {
-    if (R.first->need_payload()) {
-      td::actor::create_actor<HttpRldpPayloadSender>("HttpPayloadSender(R)", std::move(R.second), id_, local_id_, adnl_,
-                                                     rldp_)
-          .release();
-    }
+    td::actor::create_actor<HttpRldpPayloadSender>("HttpPayloadSender(R)", std::move(R.second), id_, local_id_, adnl_,
+                                                   rldp_)
+        .release();
     auto f = ton::serialize_tl_object(R.first->store_tl(), true);
     promise_.set_value(std::move(f));
     stop();
   }
 
   void abort_query(td::Status error) {
     LOG(INFO) << "aborting http over rldp query: " << error;
-    promise_.set_result(create_error_response(request_->proto_version(), 502, "Bad Gateway"));
+    promise_.set_result(create_error_response(proto_version_, 502, "Bad Gateway"));
     stop();
   }
 
@@ -848,6 +847,7 @@ class RldpToTcpRequestSender : public td::actor::Actor {
 
   std::unique_ptr<ton::http::HttpRequest> request_;
   std::shared_ptr<ton::http::HttpPayload> request_payload_;
+  std::string proto_version_;
 
   td::Promise<td::BufferSlice> promise_;
 
@@ -1090,6 +1090,7 @@ class RldpHttpProxy : public td::actor::Actor {
     }
 
     rldp_ = ton::rldp::Rldp::create(adnl_.get());
+    td::actor::send_closure(rldp_, &ton::rldp::Rldp::set_default_mtu, 16 << 10);
     td::actor::send_closure(rldp_, &ton::rldp::Rldp::add_id, local_id_);
     for (auto &serv_id : server_ids_) {
       td::actor::send_closure(rldp_, &ton::rldp::Rldp::add_id, serv_id);

rldp/rldp-in.hpp

@@ -71,7 +71,7 @@ class RldpIn : public RldpImpl {
 
   void send_query(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, std::string name,
                   td::Promise<td::BufferSlice> promise, td::Timestamp timeout, td::BufferSlice data) override {
-    send_query_ex(src, dst, name, std::move(promise), timeout, std::move(data), default_mtu());
+    send_query_ex(src, dst, name, std::move(promise), timeout, std::move(data), default_mtu_);
   }
   void send_query_ex(adnl::AdnlNodeIdShort src, adnl::AdnlNodeIdShort dst, std::string name,
                      td::Promise<td::BufferSlice> promise, td::Timestamp timeout, td::BufferSlice data,
@@ -101,6 +101,10 @@ class RldpIn : public RldpImpl {
   void add_id(adnl::AdnlNodeIdShort local_id) override;
   void get_conn_ip_str(adnl::AdnlNodeIdShort l_id, adnl::AdnlNodeIdShort p_id, td::Promise<td::string> promise) override;
 
+  void set_default_mtu(td::uint64 mtu) override {
+    default_mtu_ = mtu;
+  }
+
   RldpIn(td::actor::ActorId<adnl::AdnlPeerTable> adnl) : adnl_(adnl) {
   }
 
@@ -116,6 +120,7 @@ class RldpIn : public RldpImpl {
   std::set<TransferId> lru_set_;
   RldpLru lru_;
   td::uint32 lru_size_ = 0;
+  td::uint64 default_mtu_ = adnl::Adnl::get_mtu();
 
   std::map<TransferId, td::uint64> max_size_;