The view update backlog is propagated in two ways:

1. With responses to requests:
   

   scylladb/service/storage_proxy.cc

   Lines 2411 to 2447 in 4433d2e

   	void storage_proxy::got_response(storage_proxy::response_id_type id, gms::inet_address from, std::optional<db::view::update_backlog> backlog) {
   	auto it = _response_handlers.find(id);
   	if (it != _response_handlers.end()) {
   	tracing::trace(it->second->get_trace_state(), "Got a response from /{}", from);
   	if (it->second->response(from)) {
   	remove_response_handler_entry(std::move(it)); // last one, remove entry. Will cancel expiration timer too.
   	} else {
   	it->second->check_for_early_completion();
   	}
   	}
   	maybe_update_view_backlog_of(std::move(from), std::move(backlog));
   	}
   	void storage_proxy::got_failure_response(storage_proxy::response_id_type id, gms::inet_address from, size_t count, std::optional<db::view::update_backlog> backlog, error err, std::optional<sstring> msg) {
   	auto it = _response_handlers.find(id);
   	if (it != _response_handlers.end()) {
   	tracing::trace(it->second->get_trace_state(), "Got {} failures from /{}", count, from);
   	if (it->second->failure_response(from, count, err, std::move(msg))) {
   	remove_response_handler_entry(std::move(it));
   	} else {
   	it->second->check_for_early_completion();
   	}
   	}
   	maybe_update_view_backlog_of(std::move(from), std::move(backlog));
   	}
   	void storage_proxy::maybe_update_view_backlog_of(gms::inet_address replica, std::optional<db::view::update_backlog> backlog) {
   	if (backlog) {
   	auto now = std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
   	_view_update_backlogs[replica] = {std::move(*backlog), now};
   	}
   	}
   	db::view::update_backlog storage_proxy::get_view_update_backlog() const {
   	return _max_view_update_backlog.add_fetch(this_shard_id(), get_db().local().get_view_update_backlog());
   	}

2. Every gms::gossiper::INTERVAL (1s) through view_update_backlog_broker using gossip:
   

   scylladb/service/misc_services.cc

   Lines 237 to 263 in 4433d2e

   	future<> view_update_backlog_broker::start() {
   	_gossiper.register_(shared_from_this());
   	if (this_shard_id() == 0) {
   	// Gossiper runs only on shard 0, and there's no API to add multiple, per-shard application states.
   	// Also, right now we aggregate all backlogs, since the coordinator doesn't keep per-replica shard backlogs.
   	_started = seastar::async([this] {
   	std::optional<db::view::update_backlog> backlog_published;
   	while (!_as.abort_requested()) {
   	auto backlog = _sp.local().get_view_update_backlog();
   	if (backlog_published && *backlog_published == backlog) {
   	sleep_abortable(gms::gossiper::INTERVAL, _as).get();
   	continue;
   	}
   	auto now = api::timestamp_type(std::chrono::duration_cast<std::chrono::milliseconds>(
   	std::chrono::system_clock::now().time_since_epoch()).count());
   	//FIXME: discarded future.
   	(void)_gossiper.add_local_application_state(
   	gms::application_state::VIEW_BACKLOG,
   	gms::versioned_value(seastar::format("{}:{}:{}", backlog.current, backlog.max, now)));
   	backlog_published = backlog;
   	sleep_abortable(gms::gossiper::INTERVAL, _as).get();
   	}
   	}).handle_exception_type([] (const seastar::sleep_aborted& ignored) { });
   	}
   	return make_ready_future<>();
   	}

Consider the following scenario:

1. Cluster starts, all nodes gossip their empty view update backlog to one another
2. On node N, view_update_backlog_broker performs an iteration of its backlog update loop, sees no change (backlog has been empty since the start), schedules the next iteration after 1s
3. Within the next 1s, coordinator (different than N) sends a write to N causing a remote view update (which we do not wait for). As a result, node N replies immediately with an increased view update backlog, which is then noted by the coordinator.
4. Still within the 1s, node N finishes the view update in the background, dropping its view update backlog to 0.
5. In the following iterations of view_update_backlog_broker on N, backlog is empty, so no change is seen and no update is sent

After this scenario happens, the coordinator stores an information about an increased view update backlog on N even though it's actually empty

This can be fixed either by updating the last sent value backlog_published stored by view_update_backlog_broker also when handling responses, or by sending backlog updates even when we didn't notice a change.