2271300743

Committed 19 Jan 2026 11:42AM UTC coverage: 81.376% (+1.7%) from 79.701%

Build # 2271300743

Build Type

push

gitlab-ci

Committed by

web-flow

Commit Message

Merge pull request #1879 from lluiscampos/MEN-8687-ci-debian-updates

MEN-8687: Update Debian base images for CI jobs

Run Details

8791 of 10803 relevant lines covered (81.38%)

20310.08 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.99

/src/common/http/http_resumer.cpp

// Copyright 2023 Northern.tech AS
//
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
//
//        http://www.apache.org/licenses/LICENSE-2.0
//
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.

#include <common/http_resumer.hpp>

#include <regex>

#include <common/common.hpp>
#include <common/expected.hpp>

namespace mender {
namespace common {
namespace http {
namespace resumer {

namespace common = mender::common;
namespace expected = mender::common::expected;
namespace http = mender::common::http;

// Represents the parts of a Content-Range HTTP header
// See https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Range
struct RangeHeader {
        long long int range_start {0};
        long long int range_end {0};
        long long int size {0};
};
using ExpectedRangeHeader = expected::expected<RangeHeader, error::Error>;

// Parses the HTTP Content-Range header
// For an alternative implementation without regex dependency see:
// https://github.com/mendersoftware/mender/pull/1372/commits/ea711fc4dafa943266e9013fd6704da3d4518a27
ExpectedRangeHeader ParseRangeHeader(string header) {
        RangeHeader range_header {};

        std::regex content_range_regexp {R"(bytes\s+(\d+)\s?-\s?(\d+)\s?\/?\s?(\d+|\*)?)"};

        std::smatch range_matches;
        if (!regex_match(header, range_matches, content_range_regexp)) {
                return expected::unexpected(http::MakeError(
                        http::NoSuchHeaderError, "Invalid Content-Range returned from server: " + header));
        }

        auto exp_range_start = common::StringToLongLong(range_matches[1].str());
        auto exp_range_end = common::StringToLongLong(range_matches[2].str());
        if (!exp_range_start || !exp_range_end) {
                return expected::unexpected(http::MakeError(
                        http::NoSuchHeaderError, "Content-Range contains invalid number: " + header));
        }
        range_header.range_start = exp_range_start.value();
        range_header.range_end = exp_range_end.value();

        if (range_header.range_start > range_header.range_end) {
                return expected::unexpected(http::MakeError(
                        http::NoSuchHeaderError, "Invalid Content-Range returned from server: " + header));
        }

        if ((range_matches[3].matched) && (range_matches[3].str() != "*")) {
                auto exp_size = common::StringToLongLong(range_matches[3].str());
                if (!exp_size) {
                        return expected::unexpected(http::MakeError(
                                http::NoSuchHeaderError, "Content-Range contains invalid number: " + header));
                }
                range_header.size = exp_size.value();
        }

        return range_header;
}

class HeaderHandlerFunctor {
public:
        HeaderHandlerFunctor(weak_ptr<DownloadResumerClient> resumer) :
                resumer_client_ {resumer} {};

        void operator()(http::ExpectedIncomingResponsePtr exp_resp);

private:
        void HandleFirstResponse(
                const shared_ptr<DownloadResumerClient> &resumer_client,
                http::ExpectedIncomingResponsePtr exp_resp);
        void HandleNextResponse(
                const shared_ptr<DownloadResumerClient> &resumer_client,
                http::ExpectedIncomingResponsePtr exp_resp);

        weak_ptr<DownloadResumerClient> resumer_client_;
};

class BodyHandlerFunctor {
public:
        BodyHandlerFunctor(weak_ptr<DownloadResumerClient> resumer) :
                resumer_client_ {resumer} {};

        void operator()(http::ExpectedIncomingResponsePtr exp_resp);

private:
        weak_ptr<DownloadResumerClient> resumer_client_;
};

void HeaderHandlerFunctor::operator()(http::ExpectedIncomingResponsePtr exp_resp) {
        auto resumer_client = resumer_client_.lock();
        if (resumer_client) {
                // If an error has already occurred, schedule the next AsyncCall directly
                if (!exp_resp) {
                        resumer_client->logger_.Warning(exp_resp.error().String());
                        auto err = resumer_client->ScheduleNextResumeRequest();
                        if (err != error::NoError) {
                                resumer_client->logger_.Error(err.String());
                                resumer_client->CallUserHandler(expected::unexpected(err));
                        }
                        return;
                }

                if (resumer_client->resumer_state_->active_state == DownloadResumerActiveStatus::Resuming) {
                        HandleNextResponse(resumer_client, exp_resp);
                } else {
                        HandleFirstResponse(resumer_client, exp_resp);
                }
        }
}

void HeaderHandlerFunctor::HandleFirstResponse(
        const shared_ptr<DownloadResumerClient> &resumer_client,
        http::ExpectedIncomingResponsePtr exp_resp) {
        // The first response shall always call the user header callback. On resumable responses, we
        // create a our own incoming response and call the user header handler. On errors, we log a
        // warning and call the user handler with the original response

        auto resp = exp_resp.value();
        if (resp->GetStatusCode() != mender::common::http::StatusOK) {
                // Non-resumable response
                resumer_client->CallUserHandler(exp_resp);
                return;
        }

        auto exp_header = resp->GetHeader("Content-Length");
        if (!exp_header || exp_header.value() == "0") {
                resumer_client->logger_.Warning("Response does not contain Content-Length header");
                resumer_client->CallUserHandler(exp_resp);
                return;
        }

        auto exp_length = common::StringToLongLong(exp_header.value());
        if (!exp_length || exp_length.value() < 0) {
                resumer_client->logger_.Warning(
                        "Content-Length contains invalid number: " + exp_header.value());
                resumer_client->CallUserHandler(exp_resp);
                return;
        }

        // Resumable response
        resumer_client->resumer_state_->active_state = DownloadResumerActiveStatus::Resuming;
        resumer_client->resumer_state_->offset = 0;
        resumer_client->resumer_state_->content_length = exp_length.value();

        // Prepare a modified response and call user handler
        resumer_client->response_.reset(new http::IncomingResponse(*resumer_client, resp->cancelled_));
        resumer_client->response_->status_code_ = resp->GetStatusCode();
        resumer_client->response_->status_message_ = resp->GetStatusMessage();
        resumer_client->response_->headers_ = resp->GetHeaders();
        resumer_client->CallUserHandler(resumer_client->response_);
}

void HeaderHandlerFunctor::HandleNextResponse(
        const shared_ptr<DownloadResumerClient> &resumer_client,
        http::ExpectedIncomingResponsePtr exp_resp) {
        // If an error occurs during handling here, cancel the resuming and call the user handler.

        auto resp = exp_resp.value();
        auto resumer_reader = resumer_client->resumer_reader_.lock();
        if (!resumer_reader) {
                // Errors should already have been handled as part of the Cancel() inside the
                // destructor of the reader.
                return;
        }

        auto exp_content_range = resp->GetHeader("Content-Range").and_then(ParseRangeHeader);
        if (!exp_content_range) {
                resumer_client->logger_.Error(exp_content_range.error().String());
                resumer_client->CallUserHandler(expected::unexpected(exp_content_range.error()));
                return;
        }

        auto content_range = exp_content_range.value();
        if (content_range.size != 0
                && content_range.size != resumer_client->resumer_state_->content_length) {
                auto size_changed_err = http::MakeError(
                        http::DownloadResumerError,
                        "Size of artifact changed after download was resumed (expected "
                                + to_string(resumer_client->resumer_state_->content_length) + ", got "
                                + to_string(content_range.size) + ")");
                resumer_client->logger_.Error(size_changed_err.String());
                resumer_client->CallUserHandler(expected::unexpected(size_changed_err));
                return;
        }

        if ((content_range.range_end != resumer_client->resumer_state_->content_length - 1)
                || (content_range.range_start != resumer_client->resumer_state_->offset)) {
                auto bad_range_err = http::MakeError(
                        http::DownloadResumerError,
                        "HTTP server returned an different range than requested. Requested "
                                + to_string(resumer_client->resumer_state_->offset) + "-"
                                + to_string(resumer_client->resumer_state_->content_length - 1) + ", got "
                                + to_string(content_range.range_start) + "-" + to_string(content_range.range_end));
                resumer_client->logger_.Error(bad_range_err.String());
                resumer_client->CallUserHandler(expected::unexpected(bad_range_err));
                return;
        }

        // Get the reader for the new response
        auto exp_reader = resumer_client->client_.MakeBodyAsyncReader(resp);
        if (!exp_reader) {
                auto bad_range_err = exp_reader.error().WithContext("cannot get the reader after resume");
                resumer_client->logger_.Error(bad_range_err.String());
                resumer_client->CallUserHandler(expected::unexpected(bad_range_err));
                return;
        }
        // Update the inner reader of the user reader
        resumer_reader->inner_reader_ = exp_reader.value();

        // Resume reading reusing last user data (start, end, handler)
        auto err = resumer_reader->AsyncReadResume();
        if (err != error::NoError) {
                auto bad_read_err = err.WithContext("error reading after resume");
                resumer_client->logger_.Error(bad_read_err.String());
                resumer_client->CallUserHandler(expected::unexpected(bad_read_err));
                return;
        }
}

void BodyHandlerFunctor::operator()(http::ExpectedIncomingResponsePtr exp_resp) {
        auto resumer_client = resumer_client_.lock();
        if (!resumer_client) {
                return;
        }

        if (*resumer_client->cancelled_) {
                resumer_client->CallUserHandler(exp_resp);
                return;
        }

        if (resumer_client->resumer_state_->active_state == DownloadResumerActiveStatus::Inactive) {
                resumer_client->CallUserHandler(exp_resp);
                return;
        }

        // We resume the download if either:
        // * there is any error or
        // * successful read with status code Partial Content and there is still data missing
        const bool is_range_response =
                exp_resp && exp_resp.value()->GetStatusCode() == mender::common::http::StatusPartialContent;
        const bool is_data_missing =
                resumer_client->resumer_state_->offset < resumer_client->resumer_state_->content_length;
        if (!exp_resp || (is_range_response && is_data_missing)) {
                if (!exp_resp) {
                        auto resumer_reader = resumer_client->resumer_reader_.lock();
                        if (resumer_reader) {
                                resumer_reader->inner_reader_.reset();
                        }
                        if (exp_resp.error().code == make_error_condition(errc::operation_canceled)) {
                                // We don't want to resume cancelled requests, as these were
                                // cancelled for a reason.
                                resumer_client->CallUserHandler(exp_resp);
                                return;
                        }
                        resumer_client->logger_.Info(
                                "Will try to resume after error " + exp_resp.error().String());
                }

                auto err = resumer_client->ScheduleNextResumeRequest();
                if (err != error::NoError) {
                        resumer_client->logger_.Error(err.String());
                        resumer_client->CallUserHandler(expected::unexpected(err));
                        return;
                }
        } else {
                // Update headers with the last received server response. When resuming has taken place,
                // the user will get different headers on header and body handlers, representing (somehow)
                // what the resumer has been doing in its behalf.
                auto resp = exp_resp.value();
                resumer_client->response_->status_code_ = resp->GetStatusCode();
                resumer_client->response_->status_message_ = resp->GetStatusMessage();
                resumer_client->response_->headers_ = resp->GetHeaders();

                // Finished, call the user handler \o/
                resumer_client->logger_.Debug("Download resumed and completed successfully");
                resumer_client->CallUserHandler(resumer_client->response_);
        }
}

DownloadResumerAsyncReader::~DownloadResumerAsyncReader() {
        Cancel();
}

void DownloadResumerAsyncReader::Cancel() {
        auto resumer_client = resumer_client_.lock();
        if (!*cancelled_ && resumer_client) {
                resumer_client->Cancel();
        }
}

error::Error DownloadResumerAsyncReader::AsyncRead(
        vector<uint8_t>::iterator start, vector<uint8_t>::iterator end, io::AsyncIoHandler handler) {
        if (eof_) {
                handler(0);
                return error::NoError;
        }

        auto resumer_client = resumer_client_.lock();
        if (!resumer_client || *cancelled_) {
                return error::MakeError(
                        error::ProgrammingError,
                        "DownloadResumerAsyncReader::AsyncRead called after stream is destroyed");
        }
        // Save user parameters for further resumes of the body read
        resumer_client->last_read_ = {.start = start, .end = end, .handler = handler};
        return AsyncReadResume();
}

error::Error DownloadResumerAsyncReader::AsyncReadResume() {
        auto resumer_client = resumer_client_.lock();
        if (!resumer_client) {
                return error::MakeError(
                        error::ProgrammingError,
                        "DownloadResumerAsyncReader::AsyncReadResume called after client is destroyed");
        }
        return inner_reader_->AsyncRead(
                resumer_client->last_read_.start,
                resumer_client->last_read_.end,
                [this](io::ExpectedSize result) {
                        if (!result) {
                                logger_.Warning(
                                        "Reading error, a new request will be re-scheduled. "
                                        + result.error().String());
                        } else {
                                if (result.value() == 0) {
                                        eof_ = true;
                                }
                                resumer_state_->offset += result.value();
                                logger_.Debug("read " + to_string(result.value()) + " bytes");
                                auto resumer_client = resumer_client_.lock();
                                if (resumer_client) {
                                        resumer_client->last_read_.handler(result);
                                } else {
                                        logger_.Error(
                                                "AsyncRead finish handler called after resumer client has been destroyed.");
                                }
                        }
                });
}

DownloadResumerClient::DownloadResumerClient(
        const http::ClientConfig &config, events::EventLoop &event_loop) :
        resumer_state_ {make_shared<DownloadResumerClientState>()},
        client_(config, event_loop, "http_resumer:client"),
        logger_ {"http_resumer:client"},
        cancelled_ {make_shared<bool>(true)},
        retry_ {// By setting max interval to 1 minute, combined with default min interval of 1 minute,
                        // we effectively do not have exponential backoff and use fixed 1-minute intervals.
                        .backoff = http::ExponentialBackoff(chrono::minutes(1), config.retry_download_count),
                        .wait_timer = events::Timer(event_loop)} {
}

DownloadResumerClient::~DownloadResumerClient() {
        if (!*cancelled_) {
                logger_.Warning("DownloadResumerClient destroyed while request is still active!");
        }
        client_.Cancel();
}

error::Error DownloadResumerClient::AsyncCall(
        http::OutgoingRequestPtr req,
        http::ResponseHandler user_header_handler,
        http::ResponseHandler user_body_handler) {
        HeaderHandlerFunctor resumer_header_handler {shared_from_this()};
        BodyHandlerFunctor resumer_body_handler {shared_from_this()};

        user_request_ = req;
        user_header_handler_ = user_header_handler;
        user_body_handler_ = user_body_handler;

        if (!*cancelled_) {
                return error::Error(
                        make_error_condition(errc::operation_in_progress), "HTTP resumer call already ongoing");
        }

        *cancelled_ = false;
        retry_.backoff.Reset();
        resumer_state_->active_state = DownloadResumerActiveStatus::Inactive;
        resumer_state_->user_handlers_state = DownloadResumerUserHandlersStatus::None;
        return client_.AsyncCall(req, resumer_header_handler, resumer_body_handler);
}

io::ExpectedAsyncReaderPtr DownloadResumerClient::MakeBodyAsyncReader(
        http::IncomingResponsePtr resp) {
        auto exp_reader = client_.MakeBodyAsyncReader(resp);
        if (!exp_reader) {
                return exp_reader;
        }
        auto resumer_reader = make_shared<DownloadResumerAsyncReader>(
                exp_reader.value(), resumer_state_, cancelled_, shared_from_this());
        resumer_reader_ = resumer_reader;
        return resumer_reader;
}

http::OutgoingRequestPtr DownloadResumerClient::RemainingRangeRequest() const {
        auto range_req = make_shared<http::OutgoingRequest>(*user_request_);
        if (resumer_state_->content_length > 0) {
                range_req->SetHeader(
                        "Range",
                        "bytes=" + to_string(resumer_state_->offset) + "-"
                                + to_string(resumer_state_->content_length - 1));
        }
        return range_req;
};

error::Error DownloadResumerClient::ScheduleNextResumeRequest() {
        // In any case, make sure the previous HTTP request is cancelled.
        client_.Cancel();

        auto exp_interval = retry_.backoff.NextInterval();
        if (!exp_interval) {
                return http::MakeError(
                        http::DownloadResumerError,
                        "Giving up on resuming the download: " + exp_interval.error().String());
        }

        auto interval = exp_interval.value();
        logger_.Info(
                "Resuming download after "
                + to_string(chrono::duration_cast<chrono::seconds>(interval).count()) + " seconds. Retry "
                + to_string(retry_.backoff.CurrentIteration()) + "/"
                + to_string(retry_.backoff.TryCount()));

        HeaderHandlerFunctor resumer_next_header_handler {shared_from_this()};
        BodyHandlerFunctor resumer_next_body_handler {shared_from_this()};

        retry_.wait_timer.AsyncWait(
                interval, [this, resumer_next_header_handler, resumer_next_body_handler](error::Error err) {
                        if (err != error::NoError) {
                                auto err_user = http::MakeError(
                                        http::DownloadResumerError, "Unexpected error in wait timer: " + err.String());
                                logger_.Error(err_user.String());
                                CallUserHandler(expected::unexpected(err_user));
                                return;
                        }

                        auto next_call_err = client_.AsyncCall(
                                RemainingRangeRequest(), resumer_next_header_handler, resumer_next_body_handler);
                        if (next_call_err != error::NoError) {
                                // Schedule once more
                                auto err = ScheduleNextResumeRequest();
                                if (err != error::NoError) {
                                        logger_.Error(err.String());
                                        CallUserHandler(expected::unexpected(err));
                                }
                        }
                });

        return error::NoError;
}

void DownloadResumerClient::CallUserHandler(http::ExpectedIncomingResponsePtr exp_resp) {
        if (!exp_resp) {
                DoCancel();
        }
        if (resumer_state_->user_handlers_state == DownloadResumerUserHandlersStatus::None) {
                resumer_state_->user_handlers_state =
                        DownloadResumerUserHandlersStatus::HeaderHandlerCalled;
                user_header_handler_(exp_resp);
        } else if (
                resumer_state_->user_handlers_state
                == DownloadResumerUserHandlersStatus::HeaderHandlerCalled) {
                resumer_state_->user_handlers_state = DownloadResumerUserHandlersStatus::BodyHandlerCalled;
                DoCancel();
                user_body_handler_(exp_resp);
        } else {
                string msg;
                if (!exp_resp) {
                        msg = "error: " + exp_resp.error().String();
                } else {
                        auto &resp = exp_resp.value();
                        msg = "response: " + to_string(resp->GetStatusCode()) + " " + resp->GetStatusMessage();
                }
                logger_.Warning("Cannot call any user handler with " + msg);
        }
}

void DownloadResumerClient::Cancel() {
        DoCancel();
        client_.Cancel();
};

void DownloadResumerClient::DoCancel() {
        // Set cancel state and then make a new one. Those who are interested should have their own
        // pointer to the old one.
        *cancelled_ = true;
        cancelled_ = make_shared<bool>(true);
};

} // namespace resumer
} // namespace http
} // namespace common
} // namespace mender

1	// Copyright 2023 Northern.tech AS
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	#include <common/http_resumer.hpp>
16
17	#include <regex>
18
19	#include <common/common.hpp>
20	#include <common/expected.hpp>
21
22	namespace mender {
23	namespace common {
24	namespace http {
25	namespace resumer {
26
27	namespace common = mender::common;
28	namespace expected = mender::common::expected;
29	namespace http = mender::common::http;
30
31	// Represents the parts of a Content-Range HTTP header
32	// See https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Range
33	struct RangeHeader {
34	long long int range_start {0};
35	long long int range_end {0};
36	long long int size {0};
37	};
38	using ExpectedRangeHeader = expected::expected<RangeHeader, error::Error>;
39
40	// Parses the HTTP Content-Range header
41	// For an alternative implementation without regex dependency see:
42	// https://github.com/mendersoftware/mender/pull/1372/commits/ea711fc4dafa943266e9013fd6704da3d4518a27
43	ExpectedRangeHeader ParseRangeHeader(string header) {	90✔
44	RangeHeader range_header {};
45
46	std::regex content_range_regexp {R"(bytes\s+(\d+)\s?-\s?(\d+)\s?\/?\s?(\d+\|\*)?)"};	90✔
47
48	std::smatch range_matches;
49	if (!regex_match(header, range_matches, content_range_regexp)) {	90✔
50	return expected::unexpected(http::MakeError(	10✔
51	http::NoSuchHeaderError, "Invalid Content-Range returned from server: " + header));	30✔
52	}
53
54	auto exp_range_start = common::StringToLongLong(range_matches[1].str());	80✔
55	auto exp_range_end = common::StringToLongLong(range_matches[2].str());	160✔
56	if (!exp_range_start \|\| !exp_range_end) {	80✔
57	return expected::unexpected(http::MakeError(	×
58	http::NoSuchHeaderError, "Content-Range contains invalid number: " + header));	×
59	}
60	range_header.range_start = exp_range_start.value();	80✔
61	range_header.range_end = exp_range_end.value();	80✔
62
63	if (range_header.range_start > range_header.range_end) {	80✔
64	return expected::unexpected(http::MakeError(	1✔
65	http::NoSuchHeaderError, "Invalid Content-Range returned from server: " + header));	3✔
66	}
67
68	if ((range_matches[3].matched) && (range_matches[3].str() != "*")) {	153✔
69	auto exp_size = common::StringToLongLong(range_matches[3].str());	138✔
70	if (!exp_size) {	69✔
71	return expected::unexpected(http::MakeError(	×
72	http::NoSuchHeaderError, "Content-Range contains invalid number: " + header));	×
73	}
74	range_header.size = exp_size.value();	69✔
75	}
76
77	return range_header;
78	}	90✔
79
80	class HeaderHandlerFunctor {	787✔
81	public:
82	HeaderHandlerFunctor(weak_ptr<DownloadResumerClient> resumer) :
83	resumer_client_ {resumer} {};
84
85	void operator()(http::ExpectedIncomingResponsePtr exp_resp);
86
87	private:
88	void HandleFirstResponse(
89	const shared_ptr<DownloadResumerClient> &resumer_client,
90	http::ExpectedIncomingResponsePtr exp_resp);
91	void HandleNextResponse(
92	const shared_ptr<DownloadResumerClient> &resumer_client,
93	http::ExpectedIncomingResponsePtr exp_resp);
94
95	weak_ptr<DownloadResumerClient> resumer_client_;
96	};
97
98	class BodyHandlerFunctor {	860✔
99	public:
100	BodyHandlerFunctor(weak_ptr<DownloadResumerClient> resumer) :
101	resumer_client_ {resumer} {};
102
103	void operator()(http::ExpectedIncomingResponsePtr exp_resp);
104
105	private:
106	weak_ptr<DownloadResumerClient> resumer_client_;
107	};
108
109	void HeaderHandlerFunctor::operator()(http::ExpectedIncomingResponsePtr exp_resp) {	194✔
110	auto resumer_client = resumer_client_.lock();	194✔
111	if (resumer_client) {	194✔
112	// If an error has already occurred, schedule the next AsyncCall directly
113	if (!exp_resp) {	192✔
114	resumer_client->logger_.Warning(exp_resp.error().String());	10✔
115	auto err = resumer_client->ScheduleNextResumeRequest();	10✔
116	if (err != error::NoError) {	10✔
117	resumer_client->logger_.Error(err.String());	2✔
118	resumer_client->CallUserHandler(expected::unexpected(err));	2✔
119	}
120	return;
121	}
122
123	if (resumer_client->resumer_state_->active_state == DownloadResumerActiveStatus::Resuming) {	182✔
124	HandleNextResponse(resumer_client, exp_resp);	182✔
125	} else {
126	HandleFirstResponse(resumer_client, exp_resp);	182✔
127	}
128	}
129	}
130
131	void HeaderHandlerFunctor::HandleFirstResponse(	91✔
132	const shared_ptr<DownloadResumerClient> &resumer_client,
133	http::ExpectedIncomingResponsePtr exp_resp) {
134	// The first response shall always call the user header callback. On resumable responses, we
135	// create a our own incoming response and call the user header handler. On errors, we log a
136	// warning and call the user handler with the original response
137
138	auto resp = exp_resp.value();	91✔
139	if (resp->GetStatusCode() != mender::common::http::StatusOK) {	91✔
140	// Non-resumable response
141	resumer_client->CallUserHandler(exp_resp);	2✔
142	return;	2✔
143	}
144
145	auto exp_header = resp->GetHeader("Content-Length");	178✔
146	if (!exp_header \|\| exp_header.value() == "0") {	89✔
147	resumer_client->logger_.Warning("Response does not contain Content-Length header");	2✔
148	resumer_client->CallUserHandler(exp_resp);	1✔
149	return;	1✔
150	}
151
152	auto exp_length = common::StringToLongLong(exp_header.value());	88✔
153	if (!exp_length \|\| exp_length.value() < 0) {	88✔
154	resumer_client->logger_.Warning(	×
155	"Content-Length contains invalid number: " + exp_header.value());	×
156	resumer_client->CallUserHandler(exp_resp);	×
157	return;
158	}
159
160	// Resumable response
161	resumer_client->resumer_state_->active_state = DownloadResumerActiveStatus::Resuming;	88✔
162	resumer_client->resumer_state_->offset = 0;	88✔
163	resumer_client->resumer_state_->content_length = exp_length.value();	88✔
164
165	// Prepare a modified response and call user handler
166	resumer_client->response_.reset(new http::IncomingResponse(*resumer_client, resp->cancelled_));	176✔
167	resumer_client->response_->status_code_ = resp->GetStatusCode();	88✔
168	resumer_client->response_->status_message_ = resp->GetStatusMessage();	176✔
169	resumer_client->response_->headers_ = resp->GetHeaders();
170	resumer_client->CallUserHandler(resumer_client->response_);	176✔
171	}
172
173	void HeaderHandlerFunctor::HandleNextResponse(	91✔
174	const shared_ptr<DownloadResumerClient> &resumer_client,
175	http::ExpectedIncomingResponsePtr exp_resp) {
176	// If an error occurs during handling here, cancel the resuming and call the user handler.
177
178	auto resp = exp_resp.value();	91✔
179	auto resumer_reader = resumer_client->resumer_reader_.lock();	91✔
180	if (!resumer_reader) {	91✔
181	// Errors should already have been handled as part of the Cancel() inside the
182	// destructor of the reader.
183	return;
184	}
185
186	auto exp_content_range = resp->GetHeader("Content-Range").and_then(ParseRangeHeader);	182✔
187	if (!exp_content_range) {	91✔
188	resumer_client->logger_.Error(exp_content_range.error().String());	24✔
189	resumer_client->CallUserHandler(expected::unexpected(exp_content_range.error()));	12✔
190	return;	12✔
191	}
192
193	auto content_range = exp_content_range.value();	79✔
194	if (content_range.size != 0
195	&& content_range.size != resumer_client->resumer_state_->content_length) {	79✔
196	auto size_changed_err = http::MakeError(
197	http::DownloadResumerError,
198	"Size of artifact changed after download was resumed (expected "
199	+ to_string(resumer_client->resumer_state_->content_length) + ", got "	4✔
200	+ to_string(content_range.size) + ")");	6✔
201	resumer_client->logger_.Error(size_changed_err.String());	4✔
202	resumer_client->CallUserHandler(expected::unexpected(size_changed_err));	4✔
203	return;
204	}
205
206	if ((content_range.range_end != resumer_client->resumer_state_->content_length - 1)	77✔
207	\|\| (content_range.range_start != resumer_client->resumer_state_->offset)) {	77✔
208	auto bad_range_err = http::MakeError(
209	http::DownloadResumerError,
210	"HTTP server returned an different range than requested. Requested "
211	+ to_string(resumer_client->resumer_state_->offset) + "-"	4✔
212	+ to_string(resumer_client->resumer_state_->content_length - 1) + ", got "	6✔
213	+ to_string(content_range.range_start) + "-" + to_string(content_range.range_end));	6✔
214	resumer_client->logger_.Error(bad_range_err.String());	4✔
215	resumer_client->CallUserHandler(expected::unexpected(bad_range_err));	4✔
216	return;
217	}
218
219	// Get the reader for the new response
220	auto exp_reader = resumer_client->client_.MakeBodyAsyncReader(resp);	150✔
221	if (!exp_reader) {	75✔
222	auto bad_range_err = exp_reader.error().WithContext("cannot get the reader after resume");	×
223	resumer_client->logger_.Error(bad_range_err.String());	×
224	resumer_client->CallUserHandler(expected::unexpected(bad_range_err));	×
225	return;
226	}
227	// Update the inner reader of the user reader
228	resumer_reader->inner_reader_ = exp_reader.value();	75✔
229
230	// Resume reading reusing last user data (start, end, handler)
231	auto err = resumer_reader->AsyncReadResume();	75✔
232	if (err != error::NoError) {	75✔
233	auto bad_read_err = err.WithContext("error reading after resume");	×
234	resumer_client->logger_.Error(bad_read_err.String());	×
235	resumer_client->CallUserHandler(expected::unexpected(bad_read_err));	×
236	return;
237	}
238	}
239
240	void BodyHandlerFunctor::operator()(http::ExpectedIncomingResponsePtr exp_resp) {	182✔
241	auto resumer_client = resumer_client_.lock();	182✔
242	if (!resumer_client) {	182✔
243	return;
244	}
245
246	if (*resumer_client->cancelled_) {	182✔
247	resumer_client->CallUserHandler(exp_resp);	25✔
248	return;	25✔
249	}
250
251	if (resumer_client->resumer_state_->active_state == DownloadResumerActiveStatus::Inactive) {	157✔
252	resumer_client->CallUserHandler(exp_resp);	3✔
253	return;	3✔
254	}
255
256	// We resume the download if either:
257	// * there is any error or
258	// * successful read with status code Partial Content and there is still data missing
259	const bool is_range_response =
260	exp_resp && exp_resp.value()->GetStatusCode() == mender::common::http::StatusPartialContent;	154✔
261	const bool is_data_missing =
262	resumer_client->resumer_state_->offset < resumer_client->resumer_state_->content_length;	154✔
263	if (!exp_resp \|\| (is_range_response && is_data_missing)) {	154✔
264	if (!exp_resp) {	95✔
265	auto resumer_reader = resumer_client->resumer_reader_.lock();	95✔
266	if (resumer_reader) {	95✔
267	resumer_reader->inner_reader_.reset();	95✔
268	}
269	if (exp_resp.error().code == make_error_condition(errc::operation_canceled)) {	95✔
270	// We don't want to resume cancelled requests, as these were
271	// cancelled for a reason.
272	resumer_client->CallUserHandler(exp_resp);	×
273	return;
274	}
275	resumer_client->logger_.Info(	190✔
276	"Will try to resume after error " + exp_resp.error().String());	190✔
277	}
278
279	auto err = resumer_client->ScheduleNextResumeRequest();	95✔
280	if (err != error::NoError) {	95✔
281	resumer_client->logger_.Error(err.String());	4✔
282	resumer_client->CallUserHandler(expected::unexpected(err));	4✔
283	return;
284	}
285	} else {
286	// Update headers with the last received server response. When resuming has taken place,
287	// the user will get different headers on header and body handlers, representing (somehow)
288	// what the resumer has been doing in its behalf.
289	auto resp = exp_resp.value();	59✔
290	resumer_client->response_->status_code_ = resp->GetStatusCode();	59✔
291	resumer_client->response_->status_message_ = resp->GetStatusMessage();	118✔
292	resumer_client->response_->headers_ = resp->GetHeaders();
293
294	// Finished, call the user handler \o/
295	resumer_client->logger_.Debug("Download resumed and completed successfully");	59✔
296	resumer_client->CallUserHandler(resumer_client->response_);	118✔
297	}
298	}
299
300	DownloadResumerAsyncReader::~DownloadResumerAsyncReader() {	261✔
301	Cancel();	87✔
302	}	261✔
303
304	void DownloadResumerAsyncReader::Cancel() {	87✔
305	auto resumer_client = resumer_client_.lock();	87✔
306	if (!*cancelled_ && resumer_client) {	87✔
307	resumer_client->Cancel();	3✔
308	}
309	}	87✔
310
311	error::Error DownloadResumerAsyncReader::AsyncRead(	2,522✔
312	vector<uint8_t>::iterator start, vector<uint8_t>::iterator end, io::AsyncIoHandler handler) {
313	if (eof_) {	2,522✔
314	handler(0);	×
315	return error::NoError;	×
316	}
317
318	auto resumer_client = resumer_client_.lock();	2,522✔
319	if (!resumer_client \|\| *cancelled_) {	2,522✔
320	return error::MakeError(	×
321	error::ProgrammingError,
322	"DownloadResumerAsyncReader::AsyncRead called after stream is destroyed");	×
323	}
324	// Save user parameters for further resumes of the body read
325	resumer_client->last_read_ = {.start = start, .end = end, .handler = handler};	2,522✔
326	return AsyncReadResume();	2,522✔
327	}	2,522✔
328
329	error::Error DownloadResumerAsyncReader::AsyncReadResume() {	2,597✔
330	auto resumer_client = resumer_client_.lock();	2,597✔
331	if (!resumer_client) {	2,597✔
332	return error::MakeError(	×
333	error::ProgrammingError,
334	"DownloadResumerAsyncReader::AsyncReadResume called after client is destroyed");	×
335	}
336	return inner_reader_->AsyncRead(	5,194✔
337	resumer_client->last_read_.start,
338	resumer_client->last_read_.end,
339	[this](io::ExpectedSize result) {	7,790✔
340	if (!result) {	2,596✔
341	logger_.Warning(	190✔
342	"Reading error, a new request will be re-scheduled. "
343	+ result.error().String());	190✔
344	} else {
345	if (result.value() == 0) {	2,501✔
346	eof_ = true;	59✔
347	}
348	resumer_state_->offset += result.value();	2,501✔
349	logger_.Debug("read " + to_string(result.value()) + " bytes");	2,501✔
350	auto resumer_client = resumer_client_.lock();	2,501✔
351	if (resumer_client) {	2,501✔
352	resumer_client->last_read_.handler(result);	5,002✔
353	} else {
354	logger_.Error(	×
355	"AsyncRead finish handler called after resumer client has been destroyed.");
356	}
357	}
358	});	5,193✔
359	}
360
361	DownloadResumerClient::DownloadResumerClient(	264✔
362	const http::ClientConfig &config, events::EventLoop &event_loop) :	132✔
363	resumer_state_ {make_shared<DownloadResumerClientState>()},	132✔
364	client_(config, event_loop, "http_resumer:client"),	132✔
365	logger_ {"http_resumer:client"},	264✔
366	cancelled_ {make_shared<bool>(true)},	132✔
367	retry_ {// By setting max interval to 1 minute, combined with default min interval of 1 minute,	132✔
368	// we effectively do not have exponential backoff and use fixed 1-minute intervals.
369	.backoff = http::ExponentialBackoff(chrono::minutes(1), config.retry_download_count),
370	.wait_timer = events::Timer(event_loop)} {	396✔
371	}	264✔
372
373	DownloadResumerClient::~DownloadResumerClient() {	264✔
374	if (!*cancelled_) {	132✔
375	logger_.Warning("DownloadResumerClient destroyed while request is still active!");	4✔
376	}
377	client_.Cancel();	132✔
378	}	264✔
379
380	error::Error DownloadResumerClient::AsyncCall(	186✔
381	http::OutgoingRequestPtr req,
382	http::ResponseHandler user_header_handler,
383	http::ResponseHandler user_body_handler) {
384	HeaderHandlerFunctor resumer_header_handler {shared_from_this()};	93✔
385	BodyHandlerFunctor resumer_body_handler {shared_from_this()};	93✔
386
387	user_request_ = req;
388	user_header_handler_ = user_header_handler;	93✔
389	user_body_handler_ = user_body_handler;	93✔
390
391	if (!*cancelled_) {	93✔
392	return error::Error(
393	make_error_condition(errc::operation_in_progress), "HTTP resumer call already ongoing");	×
394	}
395
396	*cancelled_ = false;	93✔
397	retry_.backoff.Reset();
398	resumer_state_->active_state = DownloadResumerActiveStatus::Inactive;	93✔
399	resumer_state_->user_handlers_state = DownloadResumerUserHandlersStatus::None;	93✔
400	return client_.AsyncCall(req, resumer_header_handler, resumer_body_handler);	279✔
401	}
402
403	io::ExpectedAsyncReaderPtr DownloadResumerClient::MakeBodyAsyncReader(	88✔
404	http::IncomingResponsePtr resp) {
405	auto exp_reader = client_.MakeBodyAsyncReader(resp);	176✔
406	if (!exp_reader) {	88✔
407	return exp_reader;	1✔
408	}
409	auto resumer_reader = make_shared<DownloadResumerAsyncReader>(
410	exp_reader.value(), resumer_state_, cancelled_, shared_from_this());	174✔
411	resumer_reader_ = resumer_reader;
412	return resumer_reader;	87✔
413	}
414
415	http::OutgoingRequestPtr DownloadResumerClient::RemainingRangeRequest() const {	101✔
416	auto range_req = make_shared<http::OutgoingRequest>(*user_request_);	101✔
417	if (resumer_state_->content_length > 0) {	101✔
418	range_req->SetHeader(	200✔
419	"Range",
420	"bytes=" + to_string(resumer_state_->offset) + "-"	200✔
421	+ to_string(resumer_state_->content_length - 1));	300✔
422	}
423	return range_req;	101✔
424	};
425
426	error::Error DownloadResumerClient::ScheduleNextResumeRequest() {	105✔
427	// In any case, make sure the previous HTTP request is cancelled.
428	client_.Cancel();	105✔
429
430	auto exp_interval = retry_.backoff.NextInterval();	105✔
431	if (!exp_interval) {	105✔
432	return http::MakeError(
433	http::DownloadResumerError,
434	"Giving up on resuming the download: " + exp_interval.error().String());	6✔
435	}
436
437	auto interval = exp_interval.value();	102✔
438	logger_.Info(	204✔
439	"Resuming download after "
440	+ to_string(chrono::duration_cast<chrono::seconds>(interval).count()) + " seconds. Retry "	204✔
441	+ to_string(retry_.backoff.CurrentIteration()) + "/"	306✔
442	+ to_string(retry_.backoff.TryCount()));	306✔
443
444	HeaderHandlerFunctor resumer_next_header_handler {shared_from_this()};	102✔
445	BodyHandlerFunctor resumer_next_body_handler {shared_from_this()};	×
446
447	retry_.wait_timer.AsyncWait(	102✔
448	interval, [this, resumer_next_header_handler, resumer_next_body_handler](error::Error err) {	204✔
449	if (err != error::NoError) {	101✔
450	auto err_user = http::MakeError(
451	http::DownloadResumerError, "Unexpected error in wait timer: " + err.String());	×
452	logger_.Error(err_user.String());	×
453	CallUserHandler(expected::unexpected(err_user));	×
454	return;
455	}
456
457	auto next_call_err = client_.AsyncCall(	101✔
458	RemainingRangeRequest(), resumer_next_header_handler, resumer_next_body_handler);	202✔
459	if (next_call_err != error::NoError) {	101✔
460	// Schedule once more
461	auto err = ScheduleNextResumeRequest();	×
462	if (err != error::NoError) {	×
463	logger_.Error(err.String());	×
464	CallUserHandler(expected::unexpected(err));	×
465	}
466	}
467	});
468
469	return error::NoError;	102✔
470	}
471
472	void DownloadResumerClient::CallUserHandler(http::ExpectedIncomingResponsePtr exp_resp) {	197✔
473	if (!exp_resp) {	197✔
474	DoCancel();	44✔
475	}
476	if (resumer_state_->user_handlers_state == DownloadResumerUserHandlersStatus::None) {	197✔
477	resumer_state_->user_handlers_state =	91✔
478	DownloadResumerUserHandlersStatus::HeaderHandlerCalled;
479	user_header_handler_(exp_resp);	182✔
480	} else if (	106✔
481	resumer_state_->user_handlers_state
482	== DownloadResumerUserHandlersStatus::HeaderHandlerCalled) {
483	resumer_state_->user_handlers_state = DownloadResumerUserHandlersStatus::BodyHandlerCalled;	90✔
484	DoCancel();	90✔
485	user_body_handler_(exp_resp);	180✔
486	} else {
487	string msg;
488	if (!exp_resp) {	16✔
489	msg = "error: " + exp_resp.error().String();	32✔
490	} else {
491	auto &resp = exp_resp.value();	×
492	msg = "response: " + to_string(resp->GetStatusCode()) + " " + resp->GetStatusMessage();	×
493	}
494	logger_.Warning("Cannot call any user handler with " + msg);	32✔
495	}
496	}	197✔
497
498	void DownloadResumerClient::Cancel() {	11✔
499	DoCancel();	11✔
500	client_.Cancel();	11✔
501	};	11✔
502
503	void DownloadResumerClient::DoCancel() {	145✔
504	// Set cancel state and then make a new one. Those who are interested should have their own
505	// pointer to the old one.
506	*cancelled_ = true;	145✔
507	cancelled_ = make_shared<bool>(true);	145✔
508	};	145✔
509
510	} // namespace resumer
511	} // namespace http
512	} // namespace common
513	} // namespace mender

mendersoftware / mender / 2271300743

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous