Compare commits

...
Author SHA1 Message Date
Cyrille BéraudandCopilot b8df6beee0 proxy_client: give subscribe requests a deadline
A subscribe is a short request: the server answers immediately and the
device then relies on push notifications. Unlike a listen, nothing keeps
it under observation. The keepalive is only armed once the server has
answered, and no request in this client ever sets a timeout, so a
subscribe stalled in name resolution, connection or TLS handshake never
completes and never fails either.

The listener then stays registered locally while the server never learns
about it: the device stops being notified, without anything reporting an
error. This is most likely right after a wake-up, when the network stack
is still settling.

Arm a deadline on subscribe requests and fail the operation when it
expires, so restartListeners() sends the subscription again. Streaming
listens are left untouched: they are long lived by design.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-09-02 20:43:42 -04:00
2 changed files with 40 additions and 1 deletions

No files matched your search

+8
View File
@@ -10,6 +10,14 @@ namespace proxy {
constexpr const std::chrono::hours OP_TIMEOUT {24}; // one day
constexpr const std::chrono::hours OP_MARGIN {2}; // two hours
/**
* Deadline for a subscribe request to complete. Unlike a listen, a subscribe is a short
* request: the server answers immediately and the device then relies on push notifications.
* Nothing else observes it, so without a deadline a subscribe stalled in name resolution,
* connection or TLS handshake would never complete nor fail.
*/
constexpr const std::chrono::seconds SUBSCRIBE_TIMEOUT {20};
using ListenToken = uint64_t;
} // namespace proxy
+32 -1
View File
@@ -1075,6 +1075,12 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
request->set_body(body);
#endif
auto rxBuf = std::make_shared<LineSplit>();
// Completion flag shared with the subscribe deadline armed below. The request
// callbacks replace each other rather than accumulating, so the flag is raised from
// the single on_done callback instead of a dedicated one.
std::shared_ptr<std::atomic_bool> subscribeDone;
if (method != ListenMethod::LISTEN)
subscribeDone = std::make_shared<std::atomic_bool>(false);
request->add_on_body_callback([this, reqid, opstate, rxBuf, cb](const char* at, size_t length) {
try {
auto& b = *rxBuf;
@@ -1113,7 +1119,9 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
opstate->ok.store(false);
}
});
request->add_on_done_callback([this, opstate, reqid](const http::Response& response) {
request->add_on_done_callback([this, opstate, reqid, subscribeDone](const http::Response& response) {
if (subscribeDone)
subscribeDone->store(true);
if (response.status_code != 200) {
if (logger_)
logger_->error("[proxy:client] [listen] send request #{} failed with code={}",
@@ -1138,6 +1146,29 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
request->get_connection()->set_keepalive(seconds);
}
});
if (subscribeDone) {
// Give the subscribe a deadline. Nothing else would ever release it: the
// keepalive above is only armed once the server has answered, and no other
// request in this client sets a timeout. A subscribe that never completes
// leaves the listener registered locally while the server never learns about
// it, so the device stops being notified without anything reporting an error.
// Cancelling runs the on_done callback, which clears the request, and failing
// the operation lets restartListeners() send the subscription again.
std::weak_ptr<http::Request> wreq = request;
request->timeout(proxy::SUBSCRIBE_TIMEOUT,
[this, reqid, opstate, subscribeDone, wreq](const asio::error_code& ec) {
if (ec == asio::error::operation_aborted or subscribeDone->load()
or isDestroying_)
return;
if (logger_)
logger_->error("[proxy:client] [listen] subscribe request #{} timed out",
reqid);
opstate->ok.store(false);
if (auto req = wreq.lock())
req->cancel();
opFailed();
});
}
request->send();
} catch (const std::exception& e) {
if (logger_)