tdlight-team/tdlight
tdlight-team/tdlight
1
1
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Improve PTS/QTS spelling.

Browse Source
This commit is contained in:
levlam 2023-01-15 11:05:53 +03:00
parent fb935a8de2
commit 45ff12ca49
7 changed files with 89 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

84
td/telegram/MessagesManager.cpp
View File

@ -4721,7 +4721,7 @@ class GetChannelDifferenceQuery final : public Td::ResultHandler {
void on_error(Status status) final {
if (!td_->messages_manager_->on_get_dialog_error(dialog_id_, status, "GetChannelDifferenceQuery") &&
status.message() != "PERSISTENT_TIMESTAMP_INVALID") {
LOG(ERROR) << "Receive error for GetChannelDifferenceQuery for " << dialog_id_ << " with pts " << pts_
LOG(ERROR) << "Receive error for GetChannelDifferenceQuery for " << dialog_id_ << " with PTS " << pts_
<< " and limit " << limit_ << ": " << status;
}
td_->messages_manager_->on_get_channel_difference(dialog_id_, pts_, limit_, nullptr);
@ -6623,7 +6623,7 @@ void MessagesManager::skip_old_pending_pts_update(tl_object_ptr<telegram_api::Up
auto full_message_id = FullMessageId::get_full_message_id(update_new_message->message_, false);
if (update_message_ids_.count(full_message_id) > 0) {
if (new_pts == old_pts || old_pts == std::numeric_limits<int32>::max()) {
// apply sent message anyway if it is definitely non-deleted or being skipped because of pts overflow
// apply sent message anyway if it is definitely non-deleted or being skipped because of PTS overflow
auto added_full_message_id = on_get_message(std::move(update_new_message->message_), true, false, false, true,
true, "updateNewMessage with an awaited message");
if (added_full_message_id != full_message_id) {
@ -6631,8 +6631,8 @@ void MessagesManager::skip_old_pending_pts_update(tl_object_ptr<telegram_api::Up
}
return;
} else {
LOG(ERROR) << "Receive awaited sent " << full_message_id << " from " << source << " with pts " << new_pts
<< " and pts_count " << pts_count << ", but current pts is " << old_pts;
LOG(ERROR) << "Receive awaited sent " << full_message_id << " from " << source << " with PTS " << new_pts
<< " and pts_count " << pts_count << ", but current PTS is " << old_pts;
dump_debug_message_op(get_dialog(full_message_id.get_dialog_id()), 3);
}
}
@ -6641,14 +6641,14 @@ void MessagesManager::skip_old_pending_pts_update(tl_object_ptr<telegram_api::Up
auto update_sent_message = static_cast<updateSentMessage *>(update.get());
if (being_sent_messages_.count(update_sent_message->random_id_) > 0) {
if (new_pts == old_pts || old_pts == std::numeric_limits<int32>::max()) {
// apply sent message anyway if it is definitely non-deleted or being skipped because of pts overflow
// apply sent message anyway if it is definitely non-deleted or being skipped because of PTS overflow
on_send_message_success(update_sent_message->random_id_, update_sent_message->message_id_,
update_sent_message->date_, update_sent_message->ttl_period_, FileId(),
"process old updateSentMessage");
return;
} else if (update_sent_message->random_id_ != 0) {
LOG(ERROR) << "Receive awaited sent " << update_sent_message->message_id_ << " from " << source << " with pts "
<< new_pts << " and pts_count " << pts_count << ", but current pts is " << old_pts;
LOG(ERROR) << "Receive awaited sent " << update_sent_message->message_id_ << " from " << source << " with PTS "
<< new_pts << " and pts_count " << pts_count << ", but current PTS is " << old_pts;
dump_debug_message_op(get_dialog(being_sent_messages_[update_sent_message->random_id_].get_dialog_id()), 3);
}
}
@ -7866,7 +7866,7 @@ void MessagesManager::add_pending_channel_update(DialogId dialog_id, tl_object_p
return;
}
if (pts_count < 0 || new_pts <= pts_count) {
LOG(ERROR) << "Receive channel update from " << source << " with wrong pts = " << new_pts
LOG(ERROR) << "Receive channel update from " << source << " with wrong PTS = " << new_pts
<< " or pts_count = " << pts_count << ": " << oneline(to_string(update));
promise.set_value(Unit());
return;
@ -7879,7 +7879,7 @@ void MessagesManager::add_pending_channel_update(DialogId dialog_id, tl_object_p
return;
}
// TODO need to save all updates that can change result of running queries not associated with pts (for example
// TODO need to save all updates that can change result of running queries not associated with PTS (for example
// getHistory) and apply them to result of these queries
Dialog *d = get_dialog_force(dialog_id, "add_pending_channel_update 2");
@ -7894,14 +7894,14 @@ void MessagesManager::add_pending_channel_update(DialogId dialog_id, tl_object_p
}
if (new_pts <= pts && new_pts >= pts - 19999) {
LOG(INFO) << "There is no need to process an update with pts " << new_pts << " in " << dialog_id << " with pts "
LOG(INFO) << "There is no need to process an update with PTS " << new_pts << " in " << dialog_id << " with PTS "
<< pts;
promise.set_value(Unit());
return;
}
if (new_pts > pts && pts != new_pts - pts_count) {
LOG(INFO) << "Found a gap in unknown " << dialog_id << " with pts = " << pts << ". new_pts = " << new_pts
LOG(INFO) << "Found a gap in unknown " << dialog_id << " with PTS = " << pts << ". new_pts = " << new_pts
<< ", pts_count = " << pts_count << " in update from " << source;
add_postponed_channel_update(dialog_id, std::move(update), new_pts, pts_count, std::move(promise));
get_channel_difference(dialog_id, pts, true, "add_pending_channel_update 3");
@ -7966,12 +7966,12 @@ void MessagesManager::add_pending_channel_update(DialogId dialog_id, tl_object_p
}
if (old_pts != new_pts - pts_count) {
LOG(INFO) << "Found a gap in the " << dialog_id << " with pts = " << old_pts << ". new_pts = " << new_pts
LOG(INFO) << "Found a gap in the " << dialog_id << " with PTS = " << old_pts << ". new_pts = " << new_pts
<< ", pts_count = " << pts_count << " in update from " << source;
if (d->was_opened || td_->contacts_manager_->get_channel_status(channel_id).is_member() ||
is_dialog_sponsored(d)) {
add_postponed_channel_update(dialog_id, std::move(update), new_pts, pts_count, std::move(promise));
get_channel_difference(dialog_id, old_pts, true, "add_pending_channel_update pts mismatch");
get_channel_difference(dialog_id, old_pts, true, "add_pending_channel_update PTS mismatch");
} else {
promise.set_value(Unit());
}
@ -9683,7 +9683,7 @@ void MessagesManager::after_get_difference() {
for (auto &it : update_message_ids_) {
// there can be unhandled updateMessageId updates after getDifference even for ordinary chats,
// because despite updates coming during getDifference have already been applied,
// some of them could be postponed because of pts gap
// some of them could be postponed because of PTS gap
auto full_message_id = it.first;
auto dialog_id = full_message_id.get_dialog_id();
auto message_id = full_message_id.get_message_id();
@ -12646,7 +12646,7 @@ void MessagesManager::read_history_outbox(DialogId dialog_id, MessageId max_mess
return;
}
// it is impossible for just sent outgoing messages because updates are ordered by pts
// it is impossible for just sent outgoing messages because updates are ordered by PTS
if (d->last_new_message_id.is_valid() && max_message_id > d->last_new_message_id &&
dialog_id.get_type() != DialogType::Channel) {
LOG(INFO) << "Receive read outbox update about unknown " << max_message_id << " in " << dialog_id
@ -14751,7 +14751,7 @@ std::pair<DialogId, unique_ptr<MessagesManager::Message>> MessagesManager::creat
is_outgoing = supposed_to_be_outgoing;
/*
// it is useless to call getChannelDifference, because the channel pts will be increased already
// it is useless to call getChannelDifference, because the channel PTS will be increased already
if (dialog_type == DialogType::Channel && !running_get_difference_ && !running_get_channel_difference(dialog_id) &&
get_channel_difference_to_log_event_id_.count(dialog_id) == 0) {
// it is safer to completely ignore the message and re-get it through getChannelDifference
@ -15842,16 +15842,16 @@ void MessagesManager::on_get_dialogs(FolderId folder_id, vector<tl_object_ptr<te
case DialogType::User:
case DialogType::Chat:
if (has_pts) {
LOG(ERROR) << "Receive user or group " << dialog_id << " with pts";
LOG(ERROR) << "Receive user or group " << dialog_id << " with PTS";
return promise.set_error(
Status::Error(500, "Wrong query result returned: receive user or basic group chat with pts"));
Status::Error(500, "Wrong query result returned: receive user or basic group chat with PTS"));
}
break;
case DialogType::Channel:
if (!has_pts) {
LOG(ERROR) << "Receive channel " << dialog_id << " without pts";
LOG(ERROR) << "Receive channel " << dialog_id << " without PTS";
return promise.set_error(
Status::Error(500, "Wrong query result returned: receive supergroup chat without pts"));
Status::Error(500, "Wrong query result returned: receive supergroup chat without PTS"));
}
break;
case DialogType::SecretChat:
@ -23998,7 +23998,7 @@ void MessagesManager::on_get_affected_history(DialogId dialog_id, AffectedHistor
Promise<Unit> &&promise) {
TRY_STATUS_PROMISE(promise, G()->close_status());
LOG(INFO) << "Receive " << (affected_history.is_final_ ? "final " : "partial ")
<< "affected history with pts = " << affected_history.pts_
<< "affected history with PTS = " << affected_history.pts_
<< " and pts_count = " << affected_history.pts_count_;
if (affected_history.pts_count_ > 0) {
@ -27972,8 +27972,8 @@ void MessagesManager::on_message_media_edited(DialogId dialog_id, MessageId mess
// updateMessageContent was already sent and needs to be sent again,
// only if 'i' and 't' sizes from edited_content were added to the photo
auto pts = result.ok();
LOG(INFO) << "Successfully edited " << message_id << " in " << dialog_id << " with pts = " << pts
<< " and last edit pts = " << m->last_edit_pts;
LOG(INFO) << "Successfully edited " << message_id << " in " << dialog_id << " with PTS = " << pts
<< " and last edit PTS = " << m->last_edit_pts;
std::swap(m->content, m->edited_content);
bool need_send_update_message_content = m->edited_content->get_type() == MessageContentType::Photo &&
m->content->get_type() == MessageContentType::Photo;
@ -32031,7 +32031,7 @@ FullMessageId MessagesManager::on_send_message_success(int64 random_id, MessageI
auto it = being_sent_messages_.find(random_id);
if (it == being_sent_messages_.end()) {
LOG(ERROR) << "Result from sendMessage for " << new_message_id << " with random_id " << random_id << " sent at "
<< date << " comes from " << source << " after updateNewMessageId, but was not discarded by pts";
<< date << " comes from " << source << " after updateNewMessageId, but was not discarded by PTS";
return {};
}
@ -38975,7 +38975,7 @@ int32 MessagesManager::load_channel_pts(DialogId dialog_id) const {
return 0;
}
auto pts = to_integer<int32>(G()->td_db()->get_binlog_pmc()->get(get_channel_pts_key(dialog_id)));
LOG(INFO) << "Load " << dialog_id << " pts = " << pts;
LOG(INFO) << "Load " << dialog_id << " PTS = " << pts;
return pts;
}
@ -38984,7 +38984,7 @@ void MessagesManager::set_channel_pts(Dialog *d, int32 new_pts, const char *sour
CHECK(d->dialog_id.get_type() == DialogType::Channel);
LOG_IF(ERROR, running_get_channel_difference(d->dialog_id))
<< "Set pts of " << d->dialog_id << " to " << new_pts << " from " << source
<< "Set PTS of " << d->dialog_id << " to " << new_pts << " from " << source
<< " while running getChannelDifference";
if (is_debug_message_op_enabled()) {
@ -38993,7 +38993,7 @@ void MessagesManager::set_channel_pts(Dialog *d, int32 new_pts, const char *sour
// TODO delete_first_messages support in channels
if (new_pts == std::numeric_limits<int32>::max()) {
LOG(ERROR) << "Update " << d->dialog_id << " pts to -1 from " << source;
LOG(ERROR) << "Update " << d->dialog_id << " PTS to -1 from " << source;
G()->td_db()->get_binlog_pmc()->erase(get_channel_pts_key(d->dialog_id));
d->pts = std::numeric_limits<int32>::max();
if (d->pending_read_channel_inbox_pts != 0) {
@ -39001,12 +39001,12 @@ void MessagesManager::set_channel_pts(Dialog *d, int32 new_pts, const char *sour
}
return;
}
if (new_pts > d->pts || (0 < new_pts && new_pts < d->pts - 99999)) { // pts can only go up or drop cardinally
if (new_pts > d->pts || (0 < new_pts && new_pts < d->pts - 99999)) { // PTS can only go up or drop cardinally
if (new_pts < d->pts - 99999) {
LOG(WARNING) << "Pts of " << d->dialog_id << " decreases from " << d->pts << " to " << new_pts << " from "
LOG(WARNING) << "PTS of " << d->dialog_id << " decreases from " << d->pts << " to " << new_pts << " from "
<< source;
} else {
LOG(INFO) << "Update " << d->dialog_id << " pts to " << new_pts << " from " << source;
LOG(INFO) << "Update " << d->dialog_id << " PTS to " << new_pts << " from " << source;
}
d->pts = new_pts;
@ -39025,7 +39025,7 @@ void MessagesManager::set_channel_pts(Dialog *d, int32 new_pts, const char *sour
G()->td_db()->get_binlog_pmc()->set(get_channel_pts_key(d->dialog_id), to_string(new_pts));
}
} else if (new_pts < d->pts) {
LOG(ERROR) << "Receive wrong pts " << new_pts << " in " << d->dialog_id << " from " << source << ". Current pts is "
LOG(ERROR) << "Receive wrong PTS " << new_pts << " in " << d->dialog_id << " from " << source << ". Current PTS is "
<< d->pts;
}
}
@ -39181,7 +39181,7 @@ void MessagesManager::do_get_channel_difference(DialogId dialog_id, int32 pts, b
limit = MIN_CHANNEL_DIFFERENCE;
}
LOG(INFO) << "-----BEGIN GET CHANNEL DIFFERENCE----- for " << dialog_id << " with pts " << pts << " and limit "
LOG(INFO) << "-----BEGIN GET CHANNEL DIFFERENCE----- for " << dialog_id << " with PTS " << pts << " and limit "
<< limit << " from " << source;
td_->create_handler<GetChannelDifferenceQuery>()->send(dialog_id, std::move(input_channel), pts, limit, force);
@ -39503,7 +39503,7 @@ void MessagesManager::on_get_channel_difference(
channel_get_difference_retry_timeouts_.erase(dialog_id);
LOG(INFO) << "Receive result of getChannelDifference for " << dialog_id << " with pts = " << request_pts
LOG(INFO) << "Receive result of getChannelDifference for " << dialog_id << " with PTS = " << request_pts
<< " and limit = " << request_limit << " from " << source << ": " << to_string(difference_ptr);
bool have_new_messages = false;
@ -39557,18 +39557,18 @@ void MessagesManager::on_get_channel_difference(
int32 flags = difference->flags_;
is_final = (flags & CHANNEL_DIFFERENCE_FLAG_IS_FINAL) != 0;
LOG_IF(ERROR, !is_final) << "Receive channelDifferenceEmpty as result of getChannelDifference from " << source
<< " with pts = " << request_pts << " and limit = " << request_limit << " in "
<< " with PTS = " << request_pts << " and limit = " << request_limit << " in "
<< dialog_id << ", but it is not final";
if (flags & CHANNEL_DIFFERENCE_FLAG_HAS_TIMEOUT) {
timeout = difference->timeout_;
}
// bots can receive channelDifferenceEmpty with pts bigger than known pts
// bots can receive channelDifferenceEmpty with PTS bigger than known PTS
// also, this can happen for deleted channels
if (request_pts != difference->pts_ && !td_->auth_manager_->is_bot() &&
have_input_peer(dialog_id, AccessRights::Read)) {
LOG(ERROR) << "Receive channelDifferenceEmpty as result of getChannelDifference from " << source
<< " with pts = " << request_pts << " and limit = " << request_limit << " in " << dialog_id
<< " with PTS = " << request_pts << " and limit = " << request_limit << " in " << dialog_id
<< ", but PTS has changed to " << difference->pts_;
}
set_channel_pts(d, difference->pts_, "channel difference empty");
@ -39586,7 +39586,7 @@ void MessagesManager::on_get_channel_difference(
auto new_pts = difference->pts_;
if (request_pts >= new_pts && request_pts > 1 && (request_pts > new_pts || !td_->auth_manager_->is_bot())) {
LOG(ERROR) << "Receive channelDifference as result of getChannelDifference from " << source
<< " with pts = " << request_pts << " and limit = " << request_limit << " in " << dialog_id
<< " with PTS = " << request_pts << " and limit = " << request_limit << " in " << dialog_id
<< ", but PTS has changed from " << d->pts << " to " << new_pts
<< ". Difference: " << oneline(to_string(difference));
new_pts = request_pts + 1;
@ -39599,7 +39599,7 @@ void MessagesManager::on_get_channel_difference(
auto message_id = MessageId::get_message_id(message, false);
if (message_id <= cur_message_id) {
LOG(ERROR) << "Receive " << cur_message_id << " after " << message_id << " in channelDifference of "
<< dialog_id << " from " << source << " with pts " << request_pts << " and limit "
<< dialog_id << " from " << source << " with PTS " << request_pts << " and limit "
<< request_limit << ": " << to_string(difference);
after_get_channel_difference(dialog_id, false);
return;
@ -39637,7 +39637,7 @@ void MessagesManager::on_get_channel_difference(
CHECK(dialog != nullptr);
if ((dialog->flags_ & telegram_api::dialog::PTS_MASK) == 0) {
LOG(ERROR) << "Receive " << dialog_id << " without pts";
LOG(ERROR) << "Receive " << dialog_id << " without PTS";
return after_get_channel_difference(dialog_id, false);
}
@ -39650,7 +39650,7 @@ void MessagesManager::on_get_channel_difference(
auto new_pts = dialog->pts_;
if (request_pts > new_pts - request_limit) {
LOG(ERROR) << "Receive channelDifferenceTooLong as result of getChannelDifference from " << source
<< " with pts = " << request_pts << " and limit = " << request_limit << " in " << dialog_id
<< " with PTS = " << request_pts << " and limit = " << request_limit << " in " << dialog_id
<< ", but PTS has changed from " << d->pts << " to " << new_pts
<< ". Difference: " << oneline(to_string(difference));
if (request_pts >= new_pts) {
@ -39747,13 +39747,13 @@ void MessagesManager::after_get_channel_difference(DialogId dialog_id, bool succ
}
if (updates.size() != old_size || running_get_channel_difference(dialog_id)) {
if (success && update_pts - 10000 < pts && update_pts_count == 1) {
// if getChannelDifference was successful and update pts is near channel pts,
// if getChannelDifference was successful and update PTS is near channel PTS,
// we hope that the update eventually can be applied
LOG(INFO) << "Can't apply postponed channel updates";
} else {
// otherwise protect from getChannelDifference repeating calls by dropping postponed updates
LOG(WARNING) << "Failed to apply postponed updates of type " << update_id << " in " << dialog_id
<< " with pts " << pts << ", update pts is " << update_pts << ", update pts count is "
<< " with PTS " << pts << ", update PTS is " << update_pts << ", update PTS count is "
<< update_pts_count;
vector<Promise<Unit>> update_promises;
for (auto &postponed_update : updates) {

2
td/telegram/PtsManager.h
View File

@ -12,7 +12,7 @@
namespace td {
// It is not about handling gaps.
// It is about finding mem processed pts.
// It is about finding mem processed PTS.
// All checks must be done before.
class PtsManager {

8
td/telegram/SecretChatActor.cpp
View File

@ -1158,7 +1158,7 @@ void SecretChatActor::do_inbound_message_decrypted_pending(unique_ptr<log_event:
// Just save log event if necessary
auto log_event_id = message->log_event_id();
// qts
// QTS
auto qts_promise = std::move(message->promise);
if (log_event_id == 0) {
@ -1185,9 +1185,9 @@ Status SecretChatActor::do_inbound_message_decrypted(unique_ptr<log_event::Inbou
// 3. [save_log_event] => Add message to MessageManager [save_message]
// Note: if we are able to add message by random_id, we may not wait for (log event). Otherwise, we should force
// binlog flush.
// 4. [save_log_event] => Update qts [qts]
// 4. [save_log_event] => Update QTS [qts]
// 5. [save_changes; save_message; ?qts) => Remove log event [remove_log_event]
// Note: It is easier not to wait for qts. In the worst case old update will be handled again after restart.
// Note: It is easier not to wait for QTS. In the worst case old update will be handled again after restart.
auto state_id = inbound_message_states_.create();
InboundMessageState &state = *inbound_message_states_.get(state_id);
@ -1236,7 +1236,7 @@ Status SecretChatActor::do_inbound_message_decrypted(unique_ptr<log_event::Inbou
on_pfs_state_changed();
}
// qts
// QTS
auto qts_promise = std::move(message->promise);
// process message

4
td/telegram/SecretChatActor.h
View File

@ -111,7 +111,7 @@ class SecretChatActor final : public NetQueryCallback {
void cancel_chat(bool delete_history, bool is_already_discarded, Promise<> promise);
// Inbound messages
// Logevent is created by SecretChatsManager, because it must contain qts
// Logevent is created by SecretChatsManager, because it must contain QTS
void add_inbound_message(unique_ptr<log_event::InboundSecretMessage> message);
// Outbound messages
@ -479,7 +479,7 @@ class SecretChatActor final : public NetQueryCallback {
// This is completly flawed.
// (A-start_save_to_binlog ----> B-start_save_to_binlog+change_memory ----> A-finish_save_to_binlog+surprise)
//
// Instead, I suggest general solution that is already used with SeqNoState and qts
// Instead, I suggest general solution that is already used with SeqNoState and QTS
// 1. We APPLY CHANGE to memory immediately AFTER corresponding EVENT is SENT to the binlog.
// 2. We SEND CHANGE to database only after corresponding EVENT is SAVED to the binlog.
// 3. Then, we are able to ERASE EVENT just AFTER the CHANGE is SAVED to the binlog.

2
td/telegram/Td.cpp
View File

@ -3464,7 +3464,7 @@ void Td::close_impl(bool destroy_flag) {
close_flag_ = 1;
G()->set_close_flag();
send_closure(auth_manager_actor_, &AuthManager::on_closing, destroy_flag);
updates_manager_->timeout_expired(); // save pts and qts
updates_manager_->timeout_expired(); // save PTS and QTS
// wait till all request_actors will stop
request_actors_.clear();

76
td/telegram/UpdatesManager.cpp
View File

@ -250,7 +250,7 @@ void UpdatesManager::fill_pts_gap(void *td) {
min_pts = min(min_pts, updates_manager->postponed_pts_updates_.begin()->first);
max_pts = max(max_pts, updates_manager->postponed_pts_updates_.rbegin()->first);
}
string source = PSTRING() << "pts from " << updates_manager->get_pts() << " to " << min_pts << '-' << max_pts;
string source = PSTRING() << "PTS from " << updates_manager->get_pts() << " to " << min_pts << '-' << max_pts;
fill_gap(td, source.c_str());
}
@ -284,7 +284,7 @@ void UpdatesManager::fill_qts_gap(void *td) {
min_qts = updates_manager->pending_qts_updates_.begin()->first;
max_qts = updates_manager->pending_qts_updates_.rbegin()->first;
}
string source = PSTRING() << "qts from " << updates_manager->get_qts() << " to " << min_qts << '-' << max_qts;
string source = PSTRING() << "QTS from " << updates_manager->get_qts() << " to " << min_qts << '-' << max_qts;
fill_gap(td, source.c_str());
}
@ -354,7 +354,7 @@ void UpdatesManager::run_get_difference(bool is_recursive, const char *source) {
pts = 0;
}
VLOG(get_difference) << "-----BEGIN GET DIFFERENCE----- from " << source << " with pts = " << pts << ", qts = " << qts
VLOG(get_difference) << "-----BEGIN GET DIFFERENCE----- from " << source << " with PTS = " << pts << ", QTS = " << qts
<< ", date = " << date;
before_get_difference(false);
@ -471,27 +471,27 @@ void UpdatesManager::timeout_expired() {
Promise<> UpdatesManager::set_pts(int32 pts, const char *source) {
if (pts == std::numeric_limits<int32>::max()) {
LOG(WARNING) << "Update pts from " << get_pts() << " to -1 from " << source;
LOG(WARNING) << "Update PTS from " << get_pts() << " to -1 from " << source;
save_pts(pts);
auto result = add_pts(pts);
init_state();
return result;
}
Promise<> result;
if (pts > get_pts() || (0 < pts && pts < get_pts() - 399999)) { // pts can only go up or drop cardinally
if (pts > get_pts() || (0 < pts && pts < get_pts() - 399999)) { // PTS can only go up or drop cardinally
if (pts < get_pts() - 399999) {
LOG(WARNING) << "PTS decreases from " << get_pts() << " to " << pts << " from " << source;
} else {
LOG(INFO) << "Update pts from " << get_pts() << " to " << pts << " from " << source;
LOG(INFO) << "Update PTS from " << get_pts() << " to " << pts << " from " << source;
}
result = add_pts(pts);
if (last_get_difference_pts_ < get_pts() - FORCED_GET_DIFFERENCE_PTS_DIFF) {
last_get_difference_pts_ = get_pts();
schedule_get_difference("rare pts getDifference");
schedule_get_difference("rare PTS getDifference");
}
} else if (pts < get_pts() && (pts > 1 || td_->option_manager_->get_option_integer("session_count") <= 1)) {
LOG(ERROR) << "Receive wrong pts = " << pts << " from " << source << ". Current pts = " << get_pts();
LOG(ERROR) << "Receive wrong PTS = " << pts << " from " << source << ". Current PTS = " << get_pts();
}
return result;
}
@ -1070,8 +1070,8 @@ void UpdatesManager::schedule_get_difference(const char *source) {
return;
}
if (!retry_timeout_.has_timeout()) {
LOG(WARNING) << "Schedule getDifference in " << retry_time_ << " seconds with pts = " << get_pts()
<< ", qts = " << get_qts() << ", date = " << get_date() << " from " << source;
LOG(WARNING) << "Schedule getDifference in " << retry_time_ << " seconds with PTS = " << get_pts()
<< ", QTS = " << get_qts() << ", date = " << get_date() << " from " << source;
retry_timeout_.set_callback(std::move(fill_get_difference_gap));
retry_timeout_.set_callback_data(static_cast<void *>(td_));
retry_timeout_.set_timeout_in(retry_time_);
@ -1091,12 +1091,12 @@ void UpdatesManager::on_get_updates_state(tl_object_ptr<telegram_api::updates_st
// TODO use state->unread_count;
if (get_pts() == std::numeric_limits<int32>::max()) {
LOG(WARNING) << "Restore pts to " << state->pts_;
// restoring right pts
LOG(WARNING) << "Restore PTS to " << state->pts_;
// restoring right PTS
CHECK(pending_pts_updates_.empty());
auto real_running_get_difference = running_get_difference_;
running_get_difference_ = false;
process_postponed_pts_updates(); // drop all updates with old pts
process_postponed_pts_updates(); // drop all updates with old PTS
running_get_difference_ = real_running_get_difference;
pts_manager_.init(state->pts_);
last_get_difference_pts_ = get_pts();
@ -1583,7 +1583,7 @@ void UpdatesManager::on_get_difference(tl_object_ptr<telegram_api::updates_Diffe
process_pending_qts_updates();
if (!pending_qts_updates_.empty()) {
LOG(WARNING) << "Drop " << pending_qts_updates_.size() << " pending qts updates after receive empty difference";
LOG(WARNING) << "Drop " << pending_qts_updates_.size() << " pending QTS updates after receive empty difference";
auto pending_qts_updates = std::move(pending_qts_updates_);
pending_qts_updates_.clear();
@ -1747,19 +1747,19 @@ void UpdatesManager::after_get_difference() {
auto begin_time = Time::now();
auto update_count = postponed_updates.size();
VLOG(get_difference) << "Begin to apply " << postponed_updates.size()
<< " postponed pts updates with pts = " << get_pts();
<< " postponed PTS updates with PTS = " << get_pts();
for (auto &postponed_update : postponed_updates) {
auto &update = postponed_update.second;
add_pending_pts_update(std::move(update.update), update.pts, update.pts_count, update.receive_time,
std::move(update.promise), AFTER_GET_DIFFERENCE_SOURCE);
CHECK(!running_get_difference_);
}
VLOG(get_difference) << "After applying postponed pts updates have pts = " << get_pts()
VLOG(get_difference) << "After applying postponed PTS updates have PTS = " << get_pts()
<< ", max_pts = " << accumulated_pts_ << " and " << pending_pts_updates_.size() << " + "
<< postponed_pts_updates_.size() << " pending pts updates";
<< postponed_pts_updates_.size() << " pending PTS updates";
auto passed_time = Time::now() - begin_time;
if (passed_time >= UPDATE_APPLY_WARNING_TIME) {
LOG(WARNING) << "Applied " << update_count << " pts updates in " << passed_time
LOG(WARNING) << "Applied " << update_count << " PTS updates in " << passed_time
<< " seconds after postponing them for " << (Time::now() - get_difference_start_time_) << " seconds";
}
}
@ -2164,16 +2164,16 @@ void UpdatesManager::add_pending_qts_update(tl_object_ptr<telegram_api::Update>
Promise<Unit> &&promise) {
CHECK(update != nullptr);
if (qts <= 1) {
LOG(ERROR) << "Receive wrong qts " << qts << " in " << oneline(to_string(update));
schedule_get_difference("wrong qts");
LOG(ERROR) << "Receive wrong QTS " << qts << " in " << oneline(to_string(update));
schedule_get_difference("wrong QTS");
promise.set_value(Unit());
return;
}
int32 old_qts = get_qts();
LOG(INFO) << "Process update with qts = " << qts << ", current qts = " << old_qts;
LOG(INFO) << "Process update with QTS = " << qts << ", current QTS = " << old_qts;
if (qts < old_qts - 100001) {
LOG(WARNING) << "Restore qts after qts overflow from " << old_qts << " to " << qts << " by "
LOG(WARNING) << "Restore QTS after QTS overflow from " << old_qts << " to " << qts << " by "
<< oneline(to_string(update));
add_qts(qts - 1).set_value(Unit());
CHECK(get_qts() == qts - 1);
@ -2182,19 +2182,19 @@ void UpdatesManager::add_pending_qts_update(tl_object_ptr<telegram_api::Update>
}
if (qts <= old_qts) {
LOG(INFO) << "Skip already applied update with qts = " << qts;
LOG(INFO) << "Skip already applied update with QTS = " << qts;
promise.set_value(Unit());
return;
}
if (running_get_difference_ || (qts - 1 > old_qts && old_qts > 0)) {
LOG(INFO) << "Postpone update with qts = " << qts;
LOG(INFO) << "Postpone update with QTS = " << qts;
if (!running_get_difference_ && pending_qts_updates_.empty()) {
set_qts_gap_timeout(MAX_UNFILLED_GAP_TIME);
}
auto &pending_update = pending_qts_updates_[qts];
if (pending_update.update != nullptr) {
LOG(WARNING) << "Receive duplicate update with qts = " << qts;
LOG(WARNING) << "Receive duplicate update with QTS = " << qts;
} else {
pending_update.receive_time = Time::now();
}
@ -2328,11 +2328,11 @@ void UpdatesManager::process_updates(vector<tl_object_ptr<telegram_api::Update>>
void UpdatesManager::process_pts_update(tl_object_ptr<telegram_api::Update> &&update) {
CHECK(update != nullptr);
// TODO need to save all updates that can change result of running queries not associated with pts (for example
// TODO need to save all updates that can change result of running queries not associated with PTS (for example
// getHistory) and apply the updates to results of the queries
if (!check_pts_update(update)) {
LOG(ERROR) << "Receive wrong pts update: " << oneline(to_string(update));
LOG(ERROR) << "Receive wrong PTS update: " << oneline(to_string(update));
update = nullptr;
return;
}
@ -2352,16 +2352,16 @@ void UpdatesManager::add_pending_pts_update(tl_object_ptr<telegram_api::Update>
CHECK(source != nullptr);
LOG(INFO) << "Receive from " << source << " pending " << to_string(update);
if (pts_count < 0 || new_pts <= pts_count) {
LOG(ERROR) << "Receive update with wrong pts = " << new_pts << " or pts_count = " << pts_count << " from " << source
LOG(ERROR) << "Receive update with wrong PTS = " << new_pts << " or pts_count = " << pts_count << " from " << source
<< ": " << oneline(to_string(update));
return promise.set_value(Unit());
}
// TODO need to save all updates that can change result of running queries not associated with pts (for example
// TODO need to save all updates that can change result of running queries not associated with PTS (for example
// getHistory) and apply them to result of this queries
if (!check_pts_update(update)) {
LOG(ERROR) << "Receive wrong pts update from " << source << ": " << oneline(to_string(update));
LOG(ERROR) << "Receive wrong PTS update from " << source << ": " << oneline(to_string(update));
return promise.set_value(Unit());
}
@ -2376,7 +2376,7 @@ void UpdatesManager::add_pending_pts_update(tl_object_ptr<telegram_api::Update>
auto now = Time::now();
if (now > last_pts_jump_warning_time_ + 1 && (need_restore_pts || now < last_pts_jump_warning_time_ + 5) &&
!(old_pts == std::numeric_limits<int32>::max() && running_get_difference_)) {
LOG(ERROR) << "Restore pts after delete_first_messages from " << old_pts << " to " << new_pts
LOG(ERROR) << "Restore PTS after delete_first_messages from " << old_pts << " to " << new_pts
<< " is disabled, pts_count = " << pts_count << ", update is from " << source << ": "
<< oneline(to_string(update));
last_pts_jump_warning_time_ = now;
@ -2385,8 +2385,8 @@ void UpdatesManager::add_pending_pts_update(tl_object_ptr<telegram_api::Update>
set_pts_gap_timeout(0.001);
/*
LOG(WARNING) << "Restore pts after delete_first_messages";
set_pts(new_pts - 1, "restore pts after delete_first_messages");
LOG(WARNING) << "Restore PTS after delete_first_messages";
set_pts(new_pts - 1, "restore PTS after delete_first_messages");
old_pts = get_pts();
CHECK(old_pts == new_pts - 1);
*/
@ -2408,7 +2408,7 @@ void UpdatesManager::add_pending_pts_update(tl_object_ptr<telegram_api::Update>
return;
}
// is_acceptable_update check was skipped for postponed pts updates
// is_acceptable_update check was skipped for postponed PTS updates
if (source == AFTER_GET_DIFFERENCE_SOURCE && !is_acceptable_update(update.get())) {
LOG(INFO) << "Postpone again unacceptable pending update";
postpone_pts_update(std::move(update), new_pts, pts_count, receive_time, std::move(promise));
@ -2507,7 +2507,7 @@ void UpdatesManager::process_qts_update(tl_object_ptr<telegram_api::Update> &&up
LOG(DEBUG) << "Process " << to_string(update_ptr);
if (last_get_difference_qts_ < qts - FORCED_GET_DIFFERENCE_PTS_DIFF) {
if (last_get_difference_qts_ != 0) {
schedule_get_difference("rare qts getDifference");
schedule_get_difference("rare QTS getDifference");
}
last_get_difference_qts_ = qts;
}
@ -2576,7 +2576,7 @@ void UpdatesManager::process_all_pending_pts_updates() {
auto diff = Time::now() - last_pts_gap_time_;
last_pts_gap_time_ = 0;
if (diff > 0.1) {
VLOG(get_difference) << "Gap in pts from " << accumulated_pts_ - accumulated_pts_count_ << " to "
VLOG(get_difference) << "Gap in PTS from " << accumulated_pts_ - accumulated_pts_count_ << " to "
<< accumulated_pts_ << " has been filled in " << begin_diff << '-' << diff << " seconds";
}
}
@ -2631,7 +2631,7 @@ void UpdatesManager::process_postponed_pts_updates() {
if (old_pts > new_pts - pts_count || last_update_it == postponed_pts_updates_.end() ||
i == GAP_TIMEOUT_UPDATE_COUNT) {
// the updates can't be applied
VLOG(get_difference) << "Can't apply " << i << " next postponed updates with pts " << update_it->second.pts
VLOG(get_difference) << "Can't apply " << i << " next postponed updates with PTS " << update_it->second.pts
<< '-' << new_pts << ", because their pts_count is " << pts_count
<< " instead of expected " << new_pts - old_pts;
last_update_it = update_it;
@ -2795,7 +2795,7 @@ void UpdatesManager::process_pending_qts_updates() {
return;
}
LOG(DEBUG) << "Process " << pending_qts_updates_.size() << " pending qts updates";
LOG(DEBUG) << "Process " << pending_qts_updates_.size() << " pending QTS updates";
auto begin_time = Time::now();
auto initial_qts = get_qts();
int32 applied_update_count = 0;

2
td/telegram/UpdatesManager.h
View File

@ -225,7 +225,7 @@ class UpdatesManager final : public Actor {
std::multimap<int32, PendingSeqUpdates> postponed_updates_; // updates received during getDifference
std::multimap<int32, PendingSeqUpdates> pending_seq_updates_; // updates with too big seq
std::map<int32, PendingQtsUpdate> pending_qts_updates_; // updates with too big qts
std::map<int32, PendingQtsUpdate> pending_qts_updates_; // updates with too big QTS
Timeout pts_gap_timeout_;