Fix mender-update hanging when long-term network issues happen during Artifact download

src/common/events/events_io.cpp

@@ -104,27 +104,35 @@ ReaderFromAsyncReader::ReaderFromAsyncReader(EventLoop &event_loop, mio::AsyncRe
 
 mio::ExpectedSize ReaderFromAsyncReader::Read(
 	vector<uint8_t>::iterator start, vector<uint8_t>::iterator end) {
-	mio::ExpectedSize read;
-	bool finished = false;
-	event_loop_.Post([start, end, this, &finished, &read]() {
-		auto err =
-			reader_->AsyncRead(start, end, [this, &finished, &read](mio::ExpectedSize num_read) {
-				read = num_read;
-				finished = true;
-				event_loop_.Stop();
+	auto read = make_shared<mio::ExpectedSize>();
+	auto loop_stopped_here = make_shared<bool>(false);
+	event_loop_.Post([start, end, this, read, loop_stopped_here]() {
+		auto err = reader_->AsyncRead(
+			start, end, [this, read, loop_stopped_here](mio::ExpectedSize num_read) {
+				*read = num_read;
+				// Make sure to only stop the event loop here *once* because
+				// it's only recursively run once below. If this handler gets
+				// called multiple time, always stopping the loop here could
+				// stop the loop for good.
+				if (!*loop_stopped_here) {
+					event_loop_.Stop();
+					*loop_stopped_here = true;
+				}
 			});
 		if (err != error::NoError) {
-			read = expected::unexpected(err);
-			finished = true;
-			event_loop_.Stop();
+			*read = expected::unexpected(err);
+			if (!*loop_stopped_here) {
+				event_loop_.Stop();
+				*loop_stopped_here = true;
+			}
 		}
 	});
 
 	// Since the same event loop may have been used to call into this function, run the event
 	// loop recursively to keep processing events.
 	event_loop_.Run();
 
-	if (!finished) {
+	if (!*loop_stopped_here) {
 		// If this happens then it means that the event loop was stopped by somebody
 		// else. We have no choice now but to return error, since we have to get out of this
 		// stack frame. We also need to re-stop the event loop, since the first stop was
@@ -134,7 +142,7 @@ mio::ExpectedSize ReaderFromAsyncReader::Read(
 			error::Error(make_error_condition(errc::operation_canceled), "Event loop was stopped"));
 	}
 
-	return read;
+	return *read;
 }
 
 TeeReader::ExpectedTeeReaderLeafPtr TeeReader::MakeAsyncReader() {

src/common/http/http_resumer.cpp

@@ -176,7 +176,7 @@ void HeaderHandlerFunctor::HandleNextResponse(
 	// 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();
+	auto resumer_reader = resumer_client->resumer_reader_;
 	if (!resumer_reader) {
 		// Errors should already have been handled as part of the Cancel() inside the
 		// destructor of the reader.
@@ -262,7 +262,7 @@ void BodyHandlerFunctor::operator()(http::ExpectedIncomingResponsePtr exp_resp)
 		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();
+			auto resumer_reader = resumer_client->resumer_reader_;
 			if (resumer_reader) {
 				resumer_reader->inner_reader_.reset();
 			}
@@ -301,6 +301,17 @@ DownloadResumerAsyncReader::~DownloadResumerAsyncReader() {
 	Cancel();
 }
 
+void DownloadResumerAsyncReader::Fail(error::Error err) {
+	if (last_read_.handler) {
+		// Remove the handler first in case calling the handler causes this
+		// function to be called again (or another function relying on the
+		// handler).
+		auto handler = last_read_.handler;
+		last_read_.handler = nullptr;
+		handler(expected::unexpected(err));
+	}
+}
+
 void DownloadResumerAsyncReader::Cancel() {
 	auto resumer_client = resumer_client_.lock();
 	if (!*cancelled_ && resumer_client) {
@@ -322,7 +333,7 @@ error::Error DownloadResumerAsyncReader::AsyncRead(
 			"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};
+	last_read_ = {.start = start, .end = end, .handler = handler};
 	return AsyncReadResume();
 }
 
@@ -334,9 +345,7 @@ error::Error DownloadResumerAsyncReader::AsyncReadResume() {
 			"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) {
+		last_read_.start, last_read_.end, [this](io::ExpectedSize result) {
 			if (!result) {
 				logger_.Warning(
 					"Reading error, a new request will be re-scheduled. "
@@ -349,7 +358,7 @@ error::Error DownloadResumerAsyncReader::AsyncReadResume() {
 				logger_.Debug("read " + to_string(result.value()) + " bytes");
 				auto resumer_client = resumer_client_.lock();
 				if (resumer_client) {
-					resumer_client->last_read_.handler(result);
+					last_read_.handler(result);
 				} else {
 					logger_.Error(
 						"AsyncRead finish handler called after resumer client has been destroyed.");
@@ -375,6 +384,7 @@ DownloadResumerClient::~DownloadResumerClient() {
 		logger_.Warning("DownloadResumerClient destroyed while request is still active!");
 	}
 	client_.Cancel();
+	resumer_reader_.reset();
 }
 
 error::Error DownloadResumerClient::AsyncCall(
@@ -395,6 +405,7 @@ error::Error DownloadResumerClient::AsyncCall(
 
 	*cancelled_ = false;
 	retry_.backoff.Reset();
+	resumer_reader_.reset();
 	resumer_state_->active_state = DownloadResumerActiveStatus::Inactive;
 	resumer_state_->user_handlers_state = DownloadResumerUserHandlersStatus::None;
 	return client_.AsyncCall(req, resumer_header_handler, resumer_body_handler);
@@ -472,6 +483,17 @@ error::Error DownloadResumerClient::ScheduleNextResumeRequest() {
 void DownloadResumerClient::CallUserHandler(http::ExpectedIncomingResponsePtr exp_resp) {
 	if (!exp_resp) {
 		DoCancel();
+
+		// Fail the resumer_reader because that's the mechanism to deliver the
+		// information about a failure to the consumer of this
+		// (DownloadResumerClient) API handling incoming data through the body
+		// *reader* callback invoked for every chunk of data rather than by
+		// means of the body *handler* callback which is only called once full
+		// body is fetched (see the explanation of how this class works and is
+		// used near its declaration).
+		if (resumer_reader_) {
+			resumer_reader_->Fail(exp_resp.error());
+		}
 	}
 	if (resumer_state_->user_handlers_state == DownloadResumerUserHandlersStatus::None) {
 		resumer_state_->user_handlers_state =
@@ -483,6 +505,8 @@ void DownloadResumerClient::CallUserHandler(http::ExpectedIncomingResponsePtr ex
 		resumer_state_->user_handlers_state = DownloadResumerUserHandlersStatus::BodyHandlerCalled;
 		DoCancel();
 		user_body_handler_(exp_resp);
+		// we are done, the body reader won't produce any more data
+		resumer_reader_.reset();
 	} else {
 		string msg;
 		if (!exp_resp) {
@@ -498,6 +522,7 @@ void DownloadResumerClient::CallUserHandler(http::ExpectedIncomingResponsePtr ex
 void DownloadResumerClient::Cancel() {
 	DoCancel();
 	client_.Cancel();
+	resumer_reader_.reset();
 };
 
 void DownloadResumerClient::DoCancel() {

src/common/http_resumer.hpp

@@ -74,6 +74,9 @@ class DownloadResumerAsyncReader : virtual public io::AsyncReader {
 		vector<uint8_t>::iterator end,
 		io::AsyncIoHandler handler) override;
 
+	// Stop the reader with an error (unlike Cancel())
+	void Fail(error::Error err);
+
 	void Cancel() override;
 
 private:
@@ -90,6 +93,14 @@ class DownloadResumerAsyncReader : virtual public io::AsyncReader {
 
 	weak_ptr<DownloadResumerClient> resumer_client_;
 
+	// Parameters from the last time that user called AsyncRead.
+	// They are re-used when resuming the download
+	struct {
+		vector<uint8_t>::iterator start;
+		vector<uint8_t>::iterator end;
+		io::AsyncIoHandler handler;
+	} last_read_;
+
 	// The header handler needs to manipulate inner_reader_ in order to replace it in
 	// subsequent requests.
 	friend class HeaderHandlerFunctor;
@@ -99,6 +110,26 @@ class DownloadResumerAsyncReader : virtual public io::AsyncReader {
 // Main 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.
+// This is how it works:
+// - it is an http::ClientInterface class so AsyncCall() is the entry point
+// - the given header and body handlers are wrapped into handlers that
+//   transparently attempt multiple HTTP connections (and requests) in case of
+//   failures (incl. disconnections)
+// - the caller calls MakeBodyAsyncReader() in the header handler and then the
+//   result's AsyncRead() to be able to process chunks of data from potentially
+//   multiple HTTP responses, or
+// - the caller calls SetBodyWriter() in the header handler in order to
+//   capture all data from potentially multiple HTTP responses; however,
+//   SetBodyWriter() internally uses MakeBodyAsyncReader() as well, together
+//   with AsyncCopy() to move the data between the reader and the respective
+//   writer passed as argument
+// - the body handler is only called when there are no more HTTP connections and
+//   requests to be done, i.e. when all data is fetched or when an error occurs,
+//   which may happen after processing any amount of response data (from none to
+//   all but the last byte)
+// - MakeBodyAsyncReader() utilizes the above DownloadResumerAsyncReader class
+//   which then utilizes this class (DownloadResumerClient) to schedule multiple
+//   HTTP connections (and requests), if necessary, at the desired intervals
 // It needs to be used from a shared_ptr
 class DownloadResumerClient :
 	virtual public http::ClientInterface,
@@ -113,6 +144,14 @@ class DownloadResumerClient :
 		http::ResponseHandler header_handler,
 		http::ResponseHandler body_handler) override;
 
+	// NOTE: Unlike the basic http::Client which constructs the reader and
+	//       returns it, leaving the API user code to manage the reader's
+	//       lifetime, the DownloadResumerClient keeps a reference (shared
+	//       pointer) to the reader too and thus manages its lifetime. The
+	//       reason for this is that while the http::Client's reader (and it's
+	//       lifetime) is bound to a socket that gets closed on an error or
+	//       completion, only the DownloadResumerClient itself knows when the
+	//       reader won't produce more data (when no more retries).
 	io::ExpectedAsyncReaderPtr MakeBodyAsyncReader(http::IncomingResponsePtr resp) override;
 
 	void Cancel() override;
@@ -143,7 +182,7 @@ class DownloadResumerClient :
 	void DoCancel();
 
 	shared_ptr<DownloadResumerClientState> resumer_state_;
-	weak_ptr<DownloadResumerAsyncReader> resumer_reader_;
+	shared_ptr<DownloadResumerAsyncReader> resumer_reader_;
 
 	http::Client client_;
 	log::Logger logger_;
@@ -164,16 +203,6 @@ class DownloadResumerClient :
 		events::Timer wait_timer;
 	} retry_;
 
-	// Parameters from the last time that user called AsyncRead.
-	// They are re-used when resuming the download
-	struct {
-		vector<uint8_t>::iterator start;
-		vector<uint8_t>::iterator end;
-		io::AsyncIoHandler handler;
-	} last_read_;
-
-	friend class DownloadResumerAsyncReader;
-
 	friend class HeaderHandlerFunctor;
 	friend class BodyHandlerFunctor;
 };