Fix 'error 9 while decompressing xz file'

Once we've started writing data to a Sink, we can't restart a download
request, because then we end up writing duplicate data to the
Sink. Therefore we shouldn't handle retries in Downloader but at a
higher level (in particular, in copyStorePath()).

Fixes #2952.

(cherry picked from commit a67cf5a358)

src/libstore/http-binary-cache-store.cc

@@ -2,6 +2,7 @@
 #include "download.hh"
 #include "globals.hh"
 #include "nar-info-disk-cache.hh"
+#include "retry.hh"
 
 namespace nix {
 
@@ -114,7 +115,6 @@ protected:
     DownloadRequest makeRequest(const std::string & path)
     {
         DownloadRequest request(cacheUri + "/" + path);
-        request.tries = 8;
         return request;
     }
 
@@ -137,21 +137,46 @@ protected:
     {
         checkEnabled();
 
-        auto request(makeRequest(path));
+        struct State
+        {
+            DownloadRequest request;
+            std::function<void()> tryDownload;
+            unsigned int attempt = 0;
+            State(DownloadRequest && request) : request(request) {}
+        };
 
-        getDownloader()->enqueueDownload(request,
-            {[callback, this](std::future<DownloadResult> result) {
-                try {
-                    callback(result.get().data);
-                } catch (DownloadError & e) {
-                    if (e.error == Downloader::NotFound || e.error == Downloader::Forbidden)
-                        return callback(std::shared_ptr<std::string>());
-                    maybeDisable();
-                    callback.rethrow();
-                } catch (...) {
-                    callback.rethrow();
-                }
-            }});
+        auto state = std::make_shared<State>(makeRequest(path));
+
+        state->tryDownload = [callback, state, this]() {
+            getDownloader()->enqueueDownload(state->request,
+                {[callback, state, this](std::future<DownloadResult> result) {
+                    try {
+                        callback(result.get().data);
+                    } catch (DownloadError & e) {
+                        if (e.error == Downloader::NotFound || e.error == Downloader::Forbidden)
+                            return callback(std::shared_ptr<std::string>());
+                        ++state->attempt;
+                        if (state->attempt < state->request.tries && e.isTransient()) {
+                            auto ms = retrySleepTime(state->attempt);
+                            warn("%s; retrying in %d ms", e.what(), ms);
+                            /* We can't sleep here because that would
+                               block the download thread. So use a
+                               separate thread for sleeping. */
+                            std::thread([state, ms]() {
+                                std::this_thread::sleep_for(std::chrono::milliseconds(ms));
+                                state->tryDownload();
+                            }).detach();
+                        } else {
+                            maybeDisable();
+                            callback.rethrow();
+                        }
+                    } catch (...) {
+                        callback.rethrow();
+                    }
+                }});
+        };
+
+        state->tryDownload();
     }
 
 };