978727395

Committed 24 Aug 2023 11:26AM UTC coverage: 79.085% (+0.2%) from 78.84%

Build # 978727395

Build Type

push

gitlab-ci

Committed by

lluiscampos

Commit Message

feat: Implement `http::DownloadResumer`

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
`DownloadResumer` will call back these handlers _once_ (each). The data
is passed to the user at operation completion.

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

231 of 231 new or added lines in 3 files covered. (100.0%)

5706 of 7215 relevant lines covered (79.09%)

278.95 hits per line

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

82.46

/common/http/platform/beast/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.hpp>

#include <algorithm>

#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/ssl/verify_mode.hpp>
#include <boost/asio.hpp>

#include <common/common.hpp>

namespace mender {
namespace http {

namespace common = mender::common;

// 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;
        long long int range_end;
        long long int size;
};
using ExpectedRangeHeader = expected::expected<RangeHeader, error::Error>;
ExpectedRangeHeader ParseRangeHeader(string resp);

ExpectedRangeHeader ParseRangeHeader(string header) {
        RangeHeader range_header {.range_start = 0, .range_end = 0, .size = 0};

        const auto prefix_pos = string("bytes ").length();

        if (header.length() <= prefix_pos) {
                return expected::unexpected(MakeError(
                        NoSuchHeaderError, "Unexpected Content-Range received from server: " + header));
        }

        auto content = header.substr(string("bytes ").length(), header.length());

        // Split 100-200/300 into range (100-200) and size (300)
        auto range_and_size = common::SplitString(content, "/");
        if (range_and_size.size() > 2) {
                return expected::unexpected(MakeError(
                        NoSuchHeaderError, "Unexpected Content-Range received from server: " + header));
        } else if (range_and_size.size() == 2) {
                if (range_and_size[1] != "*") {
                        auto exp_size = common::StringToLongLong(range_and_size[1]);
                        if (!exp_size) {
                                return expected::unexpected(error::Error(
                                        exp_size.error().code,
                                        "Content-Range contains invalid number: " + range_and_size[1]));
                        }
                        range_header.size = exp_size.value();
                }
                content = range_and_size[0];
        }

        // Split 100-200 into range start (100) and end (200)
        auto start_and_end = common::SplitString(content, "-");
        if (start_and_end.size() != 2) {
                return expected::unexpected(
                        MakeError(NoSuchHeaderError, "Invalid Content-Range returned by server: " + content));
        }

        auto exp_range_start = common::StringToLongLong(start_and_end[0]);
        auto exp_range_end = common::StringToLongLong(start_and_end[1]);
        if (!exp_range_start || !exp_range_end) {
                return expected::unexpected(MakeError(
                        NoSuchHeaderError, "Content-Range contains invalid number: " + start_and_end[0]));
        }
        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(
                        MakeError(NoSuchHeaderError, "Invalid Content-Range returned by server: " + content));
        }

        return range_header;
}

// Base class for handler functors (header and body)
class HandlerFunctor {
protected:
        HandlerFunctor(
                ResponseHandler user_header_handler,
                ResponseHandler user_body_handler,
                OutgoingRequestPtr user_request,
                DownloadResumerClient &client) :
                user_header_handler_ {user_header_handler},
                user_body_handler_ {user_body_handler},
                user_request_ {user_request},
                client_ {client} {};

        virtual ~HandlerFunctor() = default;

        // Entrypoint. Must be implemented by child classes.
        virtual void operator()(ExpectedIncomingResponsePtr exp_resp) = 0;

        // Return a wait callback to be passed to the timer for successive calls. Must be implemented
        // by child classes.
        virtual events::EventHandler GetWaitCallback() = 0;

        // Generate a Range request from the original user request, requesting for the missing data
        // See https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Range
        OutgoingRequestPtr RemainingRangeRequest() const;

        // Schedule the next request using GetWaitCallback() from the child class
        error::Error ScheduleNextResumeRequest();

protected:
        ResponseHandler user_header_handler_;
        ResponseHandler user_body_handler_;
        OutgoingRequestPtr user_request_;
        DownloadResumerClient &client_;
};

class HeaderHandlerFunctor : virtual public HandlerFunctor {
public:
        HeaderHandlerFunctor(
                ResponseHandler user_header_handler,
                ResponseHandler user_body_handler,
                OutgoingRequestPtr user_request,
                DownloadResumerClient &client) :
                HandlerFunctor(user_header_handler, user_body_handler, user_request, client) {};

        void operator()(ExpectedIncomingResponsePtr exp_resp) override;
        events::EventHandler GetWaitCallback() override;

private:
        void HandleFirstResponse(ExpectedIncomingResponsePtr exp_resp);
        void HandleNextResponse(ExpectedIncomingResponsePtr exp_resp);
};

class BodyHandlerFunctor : virtual public HandlerFunctor {
public:
        BodyHandlerFunctor(
                ResponseHandler user_header_handler,
                ResponseHandler user_body_handler,
                OutgoingRequestPtr user_request,
                DownloadResumerClient &client) :
                HandlerFunctor(user_header_handler, user_body_handler, user_request, client) {};

        void operator()(ExpectedIncomingResponsePtr exp_resp) override;
        events::EventHandler GetWaitCallback() override;
};

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

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

        auto interval = exp_interval.value();
        client_.logger_.Info(
                "Resuming download after " + to_string(chrono::milliseconds(interval).count() / 1000)
                + " seconds");

        client_.retry_.wait_timer.AsyncWait(interval, GetWaitCallback());

        return error::NoError;
}

events::EventHandler HeaderHandlerFunctor::GetWaitCallback() {
        HeaderHandlerFunctor resumer_next_header_handler {
                user_header_handler_, user_body_handler_, user_request_, client_};
        BodyHandlerFunctor resumer_next_body_handler {
                user_header_handler_, user_body_handler_, user_request_, client_};

        return [resumer_next_header_handler,
                        resumer_next_body_handler,
                        range_req = RemainingRangeRequest()](error::Error err) {
                if (err != error::NoError) {
                        auto err_user =
                                MakeError(DownloadResumerError, "Unexpected error in wait timer: " + err.String());
                        resumer_next_header_handler.client_.logger_.Error(err_user.String());
                        resumer_next_header_handler.client_.Cancel();
                        resumer_next_header_handler.user_body_handler_(expected::unexpected(err_user));
                        return;
                }

                auto next_call_err = resumer_next_header_handler.client_.client_.AsyncCall(
                        range_req, resumer_next_header_handler, resumer_next_body_handler);
                if (next_call_err != error::NoError) {
                        auto err_user = MakeError(
                                DownloadResumerError,
                                "Failed to copy data to schedule the next resumer call: " + next_call_err.String());
                        resumer_next_header_handler.client_.logger_.Error(err_user.String());
                        resumer_next_header_handler.client_.Cancel();
                        resumer_next_header_handler.user_body_handler_(expected::unexpected(err_user));
                }
        };
}

void HeaderHandlerFunctor::operator()(ExpectedIncomingResponsePtr exp_resp) {
        if (!client_.IsOngoing()) {
                HandleFirstResponse(exp_resp);
        } else {
                HandleNextResponse(exp_resp);
        }

        if (client_.IsOngoing() && exp_resp) {
                // Set resumer BodyWriter
                auto resp = exp_resp.value();
                auto body_writer =
                        make_shared<io::ByteOffsetWriter>(client_.state_.buffer, client_.state_.buffer->size());
                body_writer.get()->SetUnlimited(true);
                resp->SetBodyWriter(body_writer);
        }
}

void HeaderHandlerFunctor::HandleFirstResponse(ExpectedIncomingResponsePtr exp_resp) {
        // The first response shall call the user header callback, both on errors and on success
        if (!exp_resp) {
                client_.logger_.Error(exp_resp.error().String());
                client_.Cancel();
                user_header_handler_(exp_resp);
                return;
        }
        auto resp = exp_resp.value();

        if (resp->GetStatusCode() != mender::http::StatusOK) {
                client_.logger_.Error("Unexpected status code " + to_string(resp->GetStatusCode()));
                client_.Cancel();
                // Return to the user the incoming response, not an error, so that it can be handled.
                user_header_handler_(exp_resp);
                return;
        }

        client_.state_.init();

        auto exp_header = resp->GetHeader("Content-Length");
        if (!exp_header || exp_header.value() == "0") {
                auto err = error::Error(
                        exp_header.error().code, "Response does not contain Content-Length header ");
                client_.logger_.Error(err.String());
                client_.Cancel();
                user_header_handler_(expected::unexpected(err));
                return;
        }

        auto exp_length = common::StringToLongLong(exp_header.value());
        if (!exp_length || exp_length.value() < 0) {
                auto err = error::Error(
                        exp_length.error().code,
                        "Content-Length contains invalid number: " + exp_header.value());
                client_.logger_.Error(err.String());
                client_.Cancel();
                user_header_handler_(expected::unexpected(err));
                return;
        }

        // Prepare state and call user handler
        client_.state_.content_length = exp_length.value();
        client_.state_.offset = 0;
        user_header_handler_(exp_resp);

        io::WriterPtr user_writer = resp->GetBodyWriter();
        if (!user_writer) {
                auto err = MakeError(
                        DownloadResumerError, "The user did not set a BodyWriter to write the data into");
                client_.logger_.Error(err.String());
                client_.Cancel();
                return;
        }
        client_.state_.user_body_writer = user_writer;
}
void HeaderHandlerFunctor::HandleNextResponse(ExpectedIncomingResponsePtr exp_resp) {
        // Subsequent responses shall not call the user handler. If an error occurs, either
        // schedule the next AsyncCall directly or save it and let the body handler forward
        // it to the user
        if (!exp_resp) {
                client_.logger_.Warning(exp_resp.error().String());

                auto err = ScheduleNextResumeRequest();
                if (err != error::NoError) {
                        client_.logger_.Error(err.String());
                        client_.Cancel();
                        user_body_handler_(expected::unexpected(err));
                }
                return;
        }
        auto resp = exp_resp.value();

        auto exp_content_range_header = resp->GetHeader("Content-Range");
        if (!exp_content_range_header) {
                client_.logger_.Error(exp_content_range_header.error().String());
                client_.Cancel();
                client_.state_.err = exp_content_range_header.error();
                return;
        }

        auto exp_content_range = ParseRangeHeader(exp_content_range_header.value());
        if (!exp_content_range) {
                client_.logger_.Error(exp_content_range.error().String());
                client_.Cancel();
                client_.state_.err = exp_content_range.error();
                return;
        }

        auto content_range = exp_content_range.value();
        if (content_range.size != 0 && content_range.size != client_.state_.content_length) {
                auto err = MakeError(
                        DownloadResumerError,
                        "Size of artifact changed after download was resumed (expected "
                                + to_string(client_.state_.content_length) + ", got "
                                + to_string(content_range.size) + ")");
                client_.logger_.Error(err.String());
                client_.Cancel();
                client_.state_.err = err;
                return;
        }
        if ((content_range.range_end != client_.state_.content_length - 1)
                || (content_range.range_start > client_.state_.offset)) {
                auto err = MakeError(
                        DownloadResumerError,
                        "HTTP server did not return expected range. Expected "
                                + to_string(client_.state_.offset) + "-"
                                + to_string(client_.state_.content_length - 1) + ", got "
                                + to_string(content_range.range_start) + "-" + to_string(content_range.range_end));
                client_.logger_.Error(err.String());
                client_.Cancel();
                client_.state_.err = err;
        } else if (content_range.range_start < client_.state_.offset) {
                // The server resumed from an earlier offset, but we can still use the data
                auto buffer = client_.state_.buffer.get();
                auto overriden = client_.state_.offset - content_range.range_start;
                client_.logger_.Warning(
                        "HTTP server returned a range with an earlier start. Expected "
                        + to_string(client_.state_.offset) + ", got " + to_string(content_range.range_start)
                        + ". The last " + to_string(overriden) + " bytes will be overriden");
                buffer->erase(buffer->end() - overriden, buffer->end());
                client_.state_.offset = buffer->size();
        }
}

events::EventHandler BodyHandlerFunctor::GetWaitCallback() {
        HeaderHandlerFunctor resumer_next_header_handler {
                user_header_handler_, user_body_handler_, user_request_, client_};
        BodyHandlerFunctor resumer_next_body_handler {
                user_header_handler_, user_body_handler_, user_request_, client_};

        return [resumer_next_header_handler,
                        resumer_next_body_handler,
                        range_req = RemainingRangeRequest()](error::Error err) {
                if (err != error::NoError) {
                        auto err_user =
                                MakeError(DownloadResumerError, "Unexpected error in wait timer: " + err.String());
                        resumer_next_body_handler.client_.logger_.Error(err_user.String());
                        resumer_next_body_handler.client_.Cancel();
                        resumer_next_body_handler.user_body_handler_(expected::unexpected(err_user));
                        return;
                }

                auto next_call_err = resumer_next_body_handler.client_.client_.AsyncCall(
                        range_req, resumer_next_header_handler, resumer_next_body_handler);
                if (next_call_err != error::NoError) {
                        auto err_user = MakeError(
                                DownloadResumerError,
                                "Failed to copy data to schedule the next resumer call: " + next_call_err.String());
                        resumer_next_body_handler.client_.logger_.Error(err_user.String());
                        resumer_next_body_handler.client_.Cancel();
                        resumer_next_body_handler.user_body_handler_(expected::unexpected(err_user));
                }
        };
}

void BodyHandlerFunctor::operator()(ExpectedIncomingResponsePtr exp_resp) {
        // It was intentionally cancelled or otherwise the header handler errored.
        if (!client_.IsOngoing()) {
                if (client_.state_.err != error::NoError) {
                        user_body_handler_(expected::unexpected(client_.state_.err));
                }
                return;
        }

        // We can resume the download if either:
        // * there is a short read (partial_message error) or
        // * successful read with status code Partial Content and there is still data missing
        const bool is_short_read =
                !exp_resp && exp_resp.error().code == http::make_error_code(http::error::partial_message);
        const bool is_range_response =
                exp_resp && exp_resp.value().get()->GetStatusCode() == mender::http::StatusPartialContent;
        const auto buffer_size = (long long int) client_.state_.buffer.get()->size();
        const bool is_data_missing = buffer_size < client_.state_.content_length;
        if (is_short_read || (is_range_response && is_data_missing)) {
                // Update resume offset
                client_.state_.offset = buffer_size;
                client_.logger_.Trace(
                        "Received " + to_string(buffer_size) + " bytes in buffer, missing "
                        + to_string(client_.state_.content_length - buffer_size) + " bytes, resuming from byte "
                        + to_string(client_.state_.offset));

                auto err = ScheduleNextResumeRequest();
                if (err != error::NoError) {
                        client_.logger_.Error(err.String());
                        client_.Cancel();
                        user_body_handler_(expected::unexpected(err));
                        return;
                }
        } else {
                // Either finished or different error. Copy data and call the user handler
                io::ByteReader body_reader {client_.state_.buffer};
                auto err = io::Copy(*client_.state_.user_body_writer.get(), body_reader);
                if (err != error::NoError) {
                        auto err_user = MakeError(
                                DownloadResumerError, "Failed to copy data to user writer: " + err.String());
                        client_.logger_.Error(err_user.String());
                        user_body_handler_(expected::unexpected(err_user));
                        return;
                }
                client_.Cancel();
                user_body_handler_(exp_resp);
        }
}

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

        return [this, req, resumer_header_handler, resumer_body_handler]() {
                return client_.AsyncCall(req, resumer_header_handler, resumer_body_handler);
        }();
}

} // namespace http
} // 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.hpp>
16
17	#include <algorithm>
18
19	#include <boost/asio/ip/tcp.hpp>
20	#include <boost/asio/ssl/verify_mode.hpp>
21	#include <boost/asio.hpp>
22
23	#include <common/common.hpp>
24
25	namespace mender {
26	namespace http {
27
28	namespace common = mender::common;
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;
34	long long int range_end;
35	long long int size;
36	};
37	using ExpectedRangeHeader = expected::expected<RangeHeader, error::Error>;
38	ExpectedRangeHeader ParseRangeHeader(string resp);
39
40	ExpectedRangeHeader ParseRangeHeader(string header) {	50✔
41	RangeHeader range_header {.range_start = 0, .range_end = 0, .size = 0};	50✔
42
43	const auto prefix_pos = string("bytes ").length();	50✔
44
45	if (header.length() <= prefix_pos) {	50✔
46	return expected::unexpected(MakeError(	3✔
47	NoSuchHeaderError, "Unexpected Content-Range received from server: " + header));	6✔
48	}
49
50	auto content = header.substr(string("bytes ").length(), header.length());	141✔
51
52	// Split 100-200/300 into range (100-200) and size (300)
53	auto range_and_size = common::SplitString(content, "/");	141✔
54	if (range_and_size.size() > 2) {	47✔
55	return expected::unexpected(MakeError(	1✔
56	NoSuchHeaderError, "Unexpected Content-Range received from server: " + header));	2✔
57	} else if (range_and_size.size() == 2) {	46✔
58	if (range_and_size[1] != "*") {	40✔
59	auto exp_size = common::StringToLongLong(range_and_size[1]);	35✔
60	if (!exp_size) {	35✔
61	return expected::unexpected(error::Error(	1✔
62	exp_size.error().code,	1✔
63	"Content-Range contains invalid number: " + range_and_size[1]));	3✔
64	}
65	range_header.size = exp_size.value();	34✔
66	}
67	content = range_and_size[0];	39✔
68	}
69
70	// Split 100-200 into range start (100) and end (200)
71	auto start_and_end = common::SplitString(content, "-");	135✔
72	if (start_and_end.size() != 2) {	45✔
73	return expected::unexpected(	2✔
74	MakeError(NoSuchHeaderError, "Invalid Content-Range returned by server: " + content));	4✔
75	}
76
77	auto exp_range_start = common::StringToLongLong(start_and_end[0]);	86✔
78	auto exp_range_end = common::StringToLongLong(start_and_end[1]);	86✔
79	if (!exp_range_start \|\| !exp_range_end) {	43✔
80	return expected::unexpected(MakeError(	2✔
81	NoSuchHeaderError, "Content-Range contains invalid number: " + start_and_end[0]));	4✔
82	}
83	range_header.range_start = exp_range_start.value();	41✔
84	range_header.range_end = exp_range_end.value();	41✔
85
86	if (range_header.range_start > range_header.range_end) {	41✔
87	return expected::unexpected(	1✔
88	MakeError(NoSuchHeaderError, "Invalid Content-Range returned by server: " + content));	2✔
89	}
90
91	return range_header;	40✔
92	}
93
94	// Base class for handler functors (header and body)
95	class HandlerFunctor {
96	protected:
97	HandlerFunctor(	228✔
98	ResponseHandler user_header_handler,
99	ResponseHandler user_body_handler,
100	OutgoingRequestPtr user_request,
101	DownloadResumerClient &client) :	228✔
102	user_header_handler_ {user_header_handler},
103	user_body_handler_ {user_body_handler},
104	user_request_ {user_request},
105	client_ {client} {};	228✔
106
107	virtual ~HandlerFunctor() = default;	1,605✔
108
109	// Entrypoint. Must be implemented by child classes.
110	virtual void operator()(ExpectedIncomingResponsePtr exp_resp) = 0;
111
112	// Return a wait callback to be passed to the timer for successive calls. Must be implemented
113	// by child classes.
114	virtual events::EventHandler GetWaitCallback() = 0;
115
116	// Generate a Range request from the original user request, requesting for the missing data
117	// See https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Range
118	OutgoingRequestPtr RemainingRangeRequest() const;
119
120	// Schedule the next request using GetWaitCallback() from the child class
121	error::Error ScheduleNextResumeRequest();
122
123	protected:
124	ResponseHandler user_header_handler_;
125	ResponseHandler user_body_handler_;
126	OutgoingRequestPtr user_request_;
127	DownloadResumerClient &client_;
128	};
129
130	class HeaderHandlerFunctor : virtual public HandlerFunctor {
131	public:
132	HeaderHandlerFunctor(	114✔
133	ResponseHandler user_header_handler,
134	ResponseHandler user_body_handler,
135	OutgoingRequestPtr user_request,
136	DownloadResumerClient &client) :	114✔
137	HandlerFunctor(user_header_handler, user_body_handler, user_request, client) {};	114✔
138
139	void operator()(ExpectedIncomingResponsePtr exp_resp) override;
140	events::EventHandler GetWaitCallback() override;
141
142	private:
143	void HandleFirstResponse(ExpectedIncomingResponsePtr exp_resp);
144	void HandleNextResponse(ExpectedIncomingResponsePtr exp_resp);
145	};
146
147	class BodyHandlerFunctor : virtual public HandlerFunctor {
148	public:
149	BodyHandlerFunctor(	114✔
150	ResponseHandler user_header_handler,
151	ResponseHandler user_body_handler,
152	OutgoingRequestPtr user_request,
153	DownloadResumerClient &client) :	114✔
154	HandlerFunctor(user_header_handler, user_body_handler, user_request, client) {};	114✔
155
156	void operator()(ExpectedIncomingResponsePtr exp_resp) override;
157	events::EventHandler GetWaitCallback() override;
158	};
159
160	OutgoingRequestPtr HandlerFunctor::RemainingRangeRequest() const {	62✔
161	auto range_req = make_shared<OutgoingRequest>(*user_request_);	62✔
162	range_req->SetHeader(	124✔
163	"Range",
164	"bytes=" + to_string(client_.state_.offset) + "-"	124✔
165	+ to_string(client_.state_.content_length - 1));	248✔
166	return range_req;	62✔
167	};
168
169	error::Error HandlerFunctor::ScheduleNextResumeRequest() {	63✔
170	auto exp_interval = client_.retry_.backoff.NextInterval();	126✔
171	if (!exp_interval) {	63✔
172	return MakeError(
173	DownloadResumerError,
174	"Giving up on resuming the download: " + exp_interval.error().String());	2✔
175	}
176
177	auto interval = exp_interval.value();	62✔
178	client_.logger_.Info(	62✔
179	"Resuming download after " + to_string(chrono::milliseconds(interval).count() / 1000)	124✔
180	+ " seconds");	124✔
181
182	client_.retry_.wait_timer.AsyncWait(interval, GetWaitCallback());	62✔
183
184	return error::NoError;	62✔
185	}
186
187	events::EventHandler HeaderHandlerFunctor::GetWaitCallback() {	10✔
188	HeaderHandlerFunctor resumer_next_header_handler {
189	user_header_handler_, user_body_handler_, user_request_, client_};	30✔
190	BodyHandlerFunctor resumer_next_body_handler {
191	user_header_handler_, user_body_handler_, user_request_, client_};	20✔
192
193	return [resumer_next_header_handler,	10✔
194	resumer_next_body_handler,
195	range_req = RemainingRangeRequest()](error::Error err) {
196	if (err != error::NoError) {	10✔
197	auto err_user =
198	MakeError(DownloadResumerError, "Unexpected error in wait timer: " + err.String());	×
199	resumer_next_header_handler.client_.logger_.Error(err_user.String());	×
200	resumer_next_header_handler.client_.Cancel();	×
201	resumer_next_header_handler.user_body_handler_(expected::unexpected(err_user));	×
202	return;	×
203	}
204
205	auto next_call_err = resumer_next_header_handler.client_.client_.AsyncCall(	10✔
206	range_req, resumer_next_header_handler, resumer_next_body_handler);	30✔
207	if (next_call_err != error::NoError) {	10✔
208	auto err_user = MakeError(
209	DownloadResumerError,
210	"Failed to copy data to schedule the next resumer call: " + next_call_err.String());	×
211	resumer_next_header_handler.client_.logger_.Error(err_user.String());	×
212	resumer_next_header_handler.client_.Cancel();	×
213	resumer_next_header_handler.user_body_handler_(expected::unexpected(err_user));	×
214	}
215	};	20✔
216	}
217
218	void HeaderHandlerFunctor::operator()(ExpectedIncomingResponsePtr exp_resp) {	114✔
219	if (!client_.IsOngoing()) {	114✔
220	HandleFirstResponse(exp_resp);	52✔
221	} else {
222	HandleNextResponse(exp_resp);	62✔
223	}
224
225	if (client_.IsOngoing() && exp_resp) {	114✔
226	// Set resumer BodyWriter
227	auto resp = exp_resp.value();	118✔
228	auto body_writer =
229	make_shared<io::ByteOffsetWriter>(client_.state_.buffer, client_.state_.buffer->size());	59✔
230	body_writer.get()->SetUnlimited(true);	59✔
231	resp->SetBodyWriter(body_writer);	59✔
232	}
233	}	114✔
234
235	void HeaderHandlerFunctor::HandleFirstResponse(ExpectedIncomingResponsePtr exp_resp) {	52✔
236	// The first response shall call the user header callback, both on errors and on success
237	if (!exp_resp) {	52✔
238	client_.logger_.Error(exp_resp.error().String());	×
239	client_.Cancel();	×
240	user_header_handler_(exp_resp);	×
241	return;	30✔
242	}
243	auto resp = exp_resp.value();	52✔
244
245	if (resp->GetStatusCode() != mender::http::StatusOK) {	52✔
246	client_.logger_.Error("Unexpected status code " + to_string(resp->GetStatusCode()));	1✔
247	client_.Cancel();	1✔
248	// Return to the user the incoming response, not an error, so that it can be handled.
249	user_header_handler_(exp_resp);	1✔
250	return;	1✔
251	}
252
253	client_.state_.init();	51✔
254
255	auto exp_header = resp->GetHeader("Content-Length");	102✔
256	if (!exp_header \|\| exp_header.value() == "0") {	51✔
257	auto err = error::Error(
258	exp_header.error().code, "Response does not contain Content-Length header ");	×
259	client_.logger_.Error(err.String());	×
260	client_.Cancel();	×
261	user_header_handler_(expected::unexpected(err));	×
262	return;	×
263	}
264
265	auto exp_length = common::StringToLongLong(exp_header.value());	51✔
266	if (!exp_length \|\| exp_length.value() < 0) {	51✔
267	auto err = error::Error(
268	exp_length.error().code,	×
269	"Content-Length contains invalid number: " + exp_header.value());	×
270	client_.logger_.Error(err.String());	×
271	client_.Cancel();	×
272	user_header_handler_(expected::unexpected(err));	×
273	return;	×
274	}
275
276	// Prepare state and call user handler
277	client_.state_.content_length = exp_length.value();	51✔
278	client_.state_.offset = 0;	51✔
279	user_header_handler_(exp_resp);	51✔
280
281	io::WriterPtr user_writer = resp->GetBodyWriter();	51✔
282	if (!user_writer) {	51✔
283	auto err = MakeError(
284	DownloadResumerError, "The user did not set a BodyWriter to write the data into");	58✔
285	client_.logger_.Error(err.String());	29✔
286	client_.Cancel();	29✔
287	return;	29✔
288	}
289	client_.state_.user_body_writer = user_writer;	22✔
290	}
291	void HeaderHandlerFunctor::HandleNextResponse(ExpectedIncomingResponsePtr exp_resp) {	62✔
292	// Subsequent responses shall not call the user handler. If an error occurs, either
293	// schedule the next AsyncCall directly or save it and let the body handler forward
294	// it to the user
295	if (!exp_resp) {	62✔
296	client_.logger_.Warning(exp_resp.error().String());	11✔
297
298	auto err = ScheduleNextResumeRequest();	11✔
299	if (err != error::NoError) {	11✔
300	client_.logger_.Error(err.String());	1✔
301	client_.Cancel();	1✔
302	user_body_handler_(expected::unexpected(err));	1✔
303	}
304	return;	11✔
305	}
306	auto resp = exp_resp.value();	51✔
307
308	auto exp_content_range_header = resp->GetHeader("Content-Range");	102✔
309	if (!exp_content_range_header) {	51✔
310	client_.logger_.Error(exp_content_range_header.error().String());	1✔
311	client_.Cancel();	1✔
312	client_.state_.err = exp_content_range_header.error();	1✔
313	return;	1✔
314	}
315
316	auto exp_content_range = ParseRangeHeader(exp_content_range_header.value());	50✔
317	if (!exp_content_range) {	50✔
318	client_.logger_.Error(exp_content_range.error().String());	10✔
319	client_.Cancel();	10✔
320	client_.state_.err = exp_content_range.error();	10✔
321	return;	10✔
322	}
323
324	auto content_range = exp_content_range.value();	40✔
325	if (content_range.size != 0 && content_range.size != client_.state_.content_length) {	40✔
326	auto err = MakeError(
327	DownloadResumerError,
328	"Size of artifact changed after download was resumed (expected "
329	+ to_string(client_.state_.content_length) + ", got "	4✔
330	+ to_string(content_range.size) + ")");	6✔
331	client_.logger_.Error(err.String());	2✔
332	client_.Cancel();	2✔
333	client_.state_.err = err;	2✔
334	return;	2✔
335	}
336	if ((content_range.range_end != client_.state_.content_length - 1)	38✔
337	\|\| (content_range.range_start > client_.state_.offset)) {	38✔
338	auto err = MakeError(
339	DownloadResumerError,
340	"HTTP server did not return expected range. Expected "
341	+ to_string(client_.state_.offset) + "-"	2✔
342	+ to_string(client_.state_.content_length - 1) + ", got "	4✔
343	+ to_string(content_range.range_start) + "-" + to_string(content_range.range_end));	5✔
344	client_.logger_.Error(err.String());	1✔
345	client_.Cancel();	1✔
346	client_.state_.err = err;	2✔
347	} else if (content_range.range_start < client_.state_.offset) {	37✔
348	// The server resumed from an earlier offset, but we can still use the data
349	auto buffer = client_.state_.buffer.get();	12✔
350	auto overriden = client_.state_.offset - content_range.range_start;	12✔
351	client_.logger_.Warning(	12✔
352	"HTTP server returned a range with an earlier start. Expected "
353	+ to_string(client_.state_.offset) + ", got " + to_string(content_range.range_start)	24✔
354	+ ". The last " + to_string(overriden) + " bytes will be overriden");	48✔
355	buffer->erase(buffer->end() - overriden, buffer->end());	12✔
356	client_.state_.offset = buffer->size();	12✔
357	}
358	}
359
360	events::EventHandler BodyHandlerFunctor::GetWaitCallback() {	52✔
361	HeaderHandlerFunctor resumer_next_header_handler {
362	user_header_handler_, user_body_handler_, user_request_, client_};	156✔
363	BodyHandlerFunctor resumer_next_body_handler {
364	user_header_handler_, user_body_handler_, user_request_, client_};	104✔
365
366	return [resumer_next_header_handler,	52✔
367	resumer_next_body_handler,
368	range_req = RemainingRangeRequest()](error::Error err) {
369	if (err != error::NoError) {	52✔
370	auto err_user =
371	MakeError(DownloadResumerError, "Unexpected error in wait timer: " + err.String());	×
372	resumer_next_body_handler.client_.logger_.Error(err_user.String());	×
373	resumer_next_body_handler.client_.Cancel();	×
374	resumer_next_body_handler.user_body_handler_(expected::unexpected(err_user));	×
375	return;	×
376	}
377
378	auto next_call_err = resumer_next_body_handler.client_.client_.AsyncCall(	52✔
379	range_req, resumer_next_header_handler, resumer_next_body_handler);	156✔
380	if (next_call_err != error::NoError) {	52✔
381	auto err_user = MakeError(
382	DownloadResumerError,
383	"Failed to copy data to schedule the next resumer call: " + next_call_err.String());	×
384	resumer_next_body_handler.client_.logger_.Error(err_user.String());	×
385	resumer_next_body_handler.client_.Cancel();	×
386	resumer_next_body_handler.user_body_handler_(expected::unexpected(err_user));	×
387	}
388	};	104✔
389	}
390
391	void BodyHandlerFunctor::operator()(ExpectedIncomingResponsePtr exp_resp) {	103✔
392	// It was intentionally cancelled or otherwise the header handler errored.
393	if (!client_.IsOngoing()) {	103✔
394	if (client_.state_.err != error::NoError) {	15✔
395	user_body_handler_(expected::unexpected(client_.state_.err));	14✔
396	}
397	return;	15✔
398	}
399
400	// We can resume the download if either:
401	// * there is a short read (partial_message error) or
402	// * successful read with status code Partial Content and there is still data missing
403	const bool is_short_read =
404	!exp_resp && exp_resp.error().code == http::make_error_code(http::error::partial_message);	88✔
405	const bool is_range_response =
406	exp_resp && exp_resp.value().get()->GetStatusCode() == mender::http::StatusPartialContent;	88✔
407	const auto buffer_size = (long long int) client_.state_.buffer.get()->size();	88✔
408	const bool is_data_missing = buffer_size < client_.state_.content_length;	88✔
409	if (is_short_read \|\| (is_range_response && is_data_missing)) {	88✔
410	// Update resume offset
411	client_.state_.offset = buffer_size;	52✔
412	client_.logger_.Trace(	52✔
413	"Received " + to_string(buffer_size) + " bytes in buffer, missing "	104✔
414	+ to_string(client_.state_.content_length - buffer_size) + " bytes, resuming from byte "	208✔
415	+ to_string(client_.state_.offset));	208✔
416
417	auto err = ScheduleNextResumeRequest();	52✔
418	if (err != error::NoError) {	52✔
419	client_.logger_.Error(err.String());	×
420	client_.Cancel();	×
421	user_body_handler_(expected::unexpected(err));	×
422	return;	×
423	}	52✔
424	} else {
425	// Either finished or different error. Copy data and call the user handler
426	io::ByteReader body_reader {client_.state_.buffer};	36✔
427	auto err = io::Copy(*client_.state_.user_body_writer.get(), body_reader);	36✔
428	if (err != error::NoError) {	36✔
429	auto err_user = MakeError(
430	DownloadResumerError, "Failed to copy data to user writer: " + err.String());	×
431	client_.logger_.Error(err_user.String());	×
432	user_body_handler_(expected::unexpected(err_user));	×
433	return;	×
434	}
435	client_.Cancel();	36✔
436	user_body_handler_(exp_resp);	36✔
437	}
438	}
439
440	error::Error DownloadResumerClient::AsyncCall(	52✔
441	OutgoingRequestPtr req,
442	ResponseHandler user_header_handler,
443	ResponseHandler user_body_handler) {
444	HeaderHandlerFunctor resumer_header_handler {
445	user_header_handler, user_body_handler, req, *this};	156✔
446	BodyHandlerFunctor resumer_body_handler {user_header_handler, user_body_handler, req, *this};	104✔
447
448	return [this, req, resumer_header_handler, resumer_body_handler]() {	52✔
449	return client_.AsyncCall(req, resumer_header_handler, resumer_body_handler);	52✔
450	}();	104✔
451	}
452
453	} // namespace http
454	} // namespace mender

mendersoftware / mender / 978727395

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