1014406969

Committed 23 Sep 2023 12:41PM UTC coverage: 77.735% (+0.2%) from 77.579%

Build # 1014406969

Build Type

push

gitlab-ci

Committed by

lluiscampos

Commit Message

feat: Implement `mender::update::http_resumer`

Implement class to download the Artifact, which will react to server
disconnections or other sorts of short read by scheduling new HTTP
requests with `Range` header.

See https://developer.mozilla.org/en-US/docs/Web/HTTP/Range_requests for
an introduction to the feature, and read the specification for more
details.

The user calls _once_ `AsyncCall` with the header and body handlers, and
`DownloadResumerClient` will call back these handlers _once_ (each). At
the user's header handler, the Reader returned by `MakeBodyAsyncReader`
is a wrapper of the actual Body reader that is taking care of the resume
of the download, calling the `AsyncRead` finalized handler only after
the download is fully completed (with potential resumes) and or/the
resuming times out.

The validation of the `Content-Range` header and the cases for the unit
tests are heavily inspired by the legacy client. See:
* https://github.com/mendersoftware/mender/blob/<a class=hub.com/mendersoftware/mender/commit/d9010526d35d3ac861ea1e4210d36c2fef748ef8">d9010526d/client/update_resumer.go#L113
* https://github.com/mendersoftware/mender/blob/d9010526d35d3ac861ea1e4210d36c2fef748ef8/client/update_resumer_test.go#L197

Ticket: MEN-6498
Changelog: None

Signed-off-by: Lluis Campos <lluis.campos@northern.tech>

Run Details

267 of 267 new or added lines in 5 files covered. (100.0%)

6686 of 8601 relevant lines covered (77.74%)

10755.08 hits per line

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

81.78

/mender-update/http_resumer/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 <mender-update/http_resumer.hpp>

#include <regex>

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

namespace mender {
namespace update {
namespace http_resumer {

namespace common = mender::common;
namespace expected = mender::common::expected;
namespace http = mender::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));
        }

        if (!range_matches[0].matched || !range_matches[1].matched) {
                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) {
                if (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 (!resumer_client->resumer_state_->initialized) {
                        HandleFirstResponse(resumer_client, exp_resp);
                } else {
                        HandleNextResponse(resumer_client, exp_resp);
                }
        }
}

void HeaderHandlerFunctor::HandleFirstResponse(
        const shared_ptr<DownloadResumerClient> &resumer_client,
        http::ExpectedIncomingResponsePtr exp_resp) {
        // The first response shall call the user header callback. On errors, we log at Warning level,
        // disable the functionality, and call the user handler with the original response

        if (!exp_resp) {
                resumer_client->DisableWithWarning(exp_resp.error().String());
                resumer_client->CallUserHandler(exp_resp);
                return;
        }
        auto resp = exp_resp.value();

        if (resp->GetStatusCode() != mender::http::StatusOK) {
                resumer_client->DisableWithWarning(
                        "Unexpected status code " + to_string(resp->GetStatusCode()));
                resumer_client->CallUserHandler(exp_resp);
                return;
        }

        auto exp_header = resp->GetHeader("Content-Length");
        if (!exp_header || exp_header.value() == "0") {
                resumer_client->DisableWithWarning("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->DisableWithWarning(
                        "Content-Length contains invalid number: " + exp_header.value());
                resumer_client->CallUserHandler(exp_resp);
                return;
        }

        // Prepare state
        resumer_client->resumer_state_ = make_shared<DownloadResumerClientState>();
        resumer_client->resumer_state_->initialized = true;
        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();
        for (auto header : resp->GetHeaders()) {
                resumer_client->response_->headers_[header.first] = header.second;
        }
        resumer_client->CallUserHandler(resumer_client->response_);
}

void HeaderHandlerFunctor::HandleNextResponse(
        const shared_ptr<DownloadResumerClient> &resumer_client,
        http::ExpectedIncomingResponsePtr exp_resp) {
        // If an error occurs has already occurred, schedule the next AsyncCall directly
        // If an error occurs during handling here, disable the resumer and call the user handler.
        if (!exp_resp) {
                resumer_client->logger_.Warning(exp_resp.error().String());

                auto err = resumer_client->ScheduleNextResumeRequest();
                if (err != error::NoError) {
                        resumer_client->DisableWithError(err);
                        resumer_client->CallUserHandler(expected::unexpected(err));
                }
                return;
        }
        auto resp = exp_resp.value();

        auto exp_content_range = resp->GetHeader("Content-Range").and_then(ParseRangeHeader);
        if (!exp_content_range) {
                resumer_client->DisableWithError(exp_content_range.error());
                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->DisableWithError(size_changed_err);
                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->DisableWithError(bad_range_err);
                resumer_client->CallUserHandler(expected::unexpected(bad_range_err));
                return;
        }

        // Get the 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->DisableWithError(bad_range_err);
                resumer_client->CallUserHandler(expected::unexpected(bad_range_err));
                return;
        } else {
                // Update the inner reader of the user reader
                resumer_client->resumer_reader_->inner_reader_ = exp_reader.value();
        }

        // Resume reading reusing last user data (start, end, handler)
        auto err = resumer_client->resumer_reader_->AsyncReadResume();
        if (err != error::NoError) {
                auto bad_read_err = err.WithContext("error reading after resume");
                resumer_client->DisableWithError(bad_read_err);
                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 || *resumer_client->cancelled_) {
                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().get()->GetStatusCode() == mender::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)) {
                auto err = resumer_client->ScheduleNextResumeRequest();
                if (err != error::NoError) {
                        resumer_client->DisableWithError(err);
                        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();
                for (auto header : resp->GetHeaders()) {
                        resumer_client->response_->headers_[header.first] = header.second;
                }

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

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

void DownloadResumerAsyncReader::Cancel() {
        if (!*cancelled_) {
                inner_reader_->Cancel();
        }
}

error::Error DownloadResumerAsyncReader::AsyncRead(
        vector<uint8_t>::iterator start, vector<uint8_t>::iterator end, io::AsyncIoHandler handler) {
        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) {
                                inner_reader_.reset();

                                logger_.Warning(
                                        "Reading error, a new request will be re-scheduled. "
                                        + result.error().String());
                        } else {
                                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) :
        client_(config, event_loop, "http_resumer:client"),
        logger_ {"http_resumer:client"},
        cancelled_ {make_shared<bool>(true)},
        retry_ {.backoff = http::ExponentialBackoff(chrono::seconds(1), 10), .wait_timer = event_loop} {
        resumer_state_ = make_shared<DownloadResumerClientState>();
}

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 call already ongoing");
        }

        *cancelled_ = false;
        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;
        }
        resumer_reader_ = make_shared<DownloadResumerAsyncReader>(
                exp_reader.value(), resumer_state_, cancelled_, shared_from_this());
        return resumer_reader_;
}

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

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

error::Error DownloadResumerClient::ScheduleNextResumeRequest() {
        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::milliseconds(interval).count() / 1000)
                + " seconds");

        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());
                                DisableWithError(err_user);
                                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) {
                                auto err_user = http::MakeError(
                                        http::DownloadResumerError,
                                        "Failed to schedule the next resumer call: " + next_call_err.String());
                                DisableWithError(err_user);
                                CallUserHandler(expected::unexpected(err_user));
                        }
                });

        return error::NoError;
}

void DownloadResumerClient::CallUserHandler(http::ExpectedIncomingResponsePtr exp_resp) {
        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;
                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::DisableWithWarning(string reason) {
        logger_.Warning(reason);
        Disable();
};

void DownloadResumerClient::DisableWithError(error::Error err) {
        logger_.Error(err.String());
        Disable();
};

void DownloadResumerClient::Disable() {
        resumer_state_->initialized = false;
        *cancelled_ = true;
        cancelled_ = make_shared<bool>(true);
        retry_.backoff.Reset();
};

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

mendersoftware / mender / 1014406969

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