Fixes for the resumable downloads feature (#2345)

* Fix handle of hashed file not closing

* Limit concurrent downloads to a maximum of 8

* Dynamically increase job size to avoid downloads appearing stalled

* Set downloader settings to avoid RAM and timeout issues

* Improve logging around downloads

* Adds more logging when starting stopping downloads
* Improves error message when GameFileSource download fails
* Stops logging errors when archive isn't available on WJ CDN

* Add retry mechanism to SingleThreadedDownloader

* Update CHANGELOG.md

* Remove hard limit for download threads

---------

Co-authored-by: UrbanCMC <UrbanCMC@web.de>

CHANGELOG.md

@@ -1,5 +1,9 @@
 ### Changelog
 
+#### Version TBD
+* Fixed issues related to high RAM usage
+  * The resumable downloads now reserve drive space to write to in advance instead of being managed in system RAM
+
 #### Version - 3.1.0.0 - 5/7/2023
 * Fixed Readme opening twice
 * Updated Text in the UI to better describe current app behavior

Wabbajack.Downloaders.Dispatcher/DownloadDispatcher.cs

@@ -5,6 +5,7 @@ using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Logging;
+using Wabbajack.Common;
 using Wabbajack.Downloaders.Interfaces;
 using Wabbajack.Downloaders.VerificationCache;
 using Wabbajack.DTOs;
@@ -50,7 +51,9 @@ public class DownloadDispatcher
 
         using var downloadScope = _logger.BeginScope("Downloading {Name}", a.Name);
         using var job = await _limiter.Begin("Downloading " + a.Name, a.Size, token);
-        return await Download(a, dest, job, token, proxy);
+        var hash = await Download(a, dest, job, token, proxy);
+        _logger.LogInformation("Finished downloading {name}. Hash: {hash}; Size: {size}/{expectedSize}", a.Name, hash, dest.Size().ToFileSizeString(), a.Size.ToFileSizeString());
+        return hash;
     }
 
     public async Task<Archive> MaybeProxy(Archive a, CancellationToken token)
@@ -153,8 +156,15 @@ public class DownloadDispatcher
         if (downloadedHash != default && (downloadedHash == archive.Hash || archive.Hash == default))
             return (DownloadResult.Success, downloadedHash);
 
-        downloadedHash = await DownloadFromMirror(archive, destination, token);
+        try
-        if (downloadedHash != default) return (DownloadResult.Mirror, downloadedHash);
+        {
+            downloadedHash = await DownloadFromMirror(archive, destination, token);
+            if (downloadedHash != default) return (DownloadResult.Mirror, downloadedHash);
+        }
+        catch (NotSupportedException)
+        {
+            // Thrown if downloading from mirror is not supported for archive, keep original hash
+        }
 
         return (DownloadResult.Failure, downloadedHash);
 
@@ -234,7 +244,7 @@ public class DownloadDispatcher
 
             return await Download(newArchive, destination, token);
         }
-        catch (Exception ex)
+        catch (Exception ex) when (ex is not NotSupportedException)
         {
             _logger.LogCritical(ex, "While finding mirror for {hash}", archive.Hash);
             return default;

Wabbajack.Installer/AInstaller.cs

@@ -422,10 +422,17 @@ public abstract class AInstaller<T>
                 await destination.Value.MoveToAsync(destination.Value.Parent.Combine(archive.Hash.ToHex()), true,
                     token);
         }
+        catch (OperationCanceledException) when (token.IsCancellationRequested)
+        {
+            // No actual error. User canceled downloads.
+        }
+        catch (NotImplementedException) when (archive.State is GameFileSource)
+        {
+            _logger.LogError("Missing game file {name}. This could be caused by missing DLC or a modified installation.", archive.Name);
+        }
         catch (Exception ex)
         {
             _logger.LogError(ex, "Download error for file {name}", archive.Name);
-            return false;
         }
 
         return false;

Wabbajack.Networking.Http/ResumableDownloader.cs

@@ -6,6 +6,7 @@ using System.Text.Json;
 using System.Threading;
 using System.Threading.Tasks;
 using Downloader;
+using Microsoft.Extensions.Logging;
 using Wabbajack.Hashing.xxHash64;
 using Wabbajack.Paths;
 using Wabbajack.Paths.IO;
@@ -19,15 +20,18 @@ internal class ResumableDownloader
     private readonly HttpRequestMessage _msg;
     private readonly AbsolutePath _outputPath;
     private readonly AbsolutePath _packagePath;
+    private readonly ILogger<SingleThreadedDownloader> _logger;
     private CancellationToken _token;
     private Exception? _error;
 
-    public ResumableDownloader(HttpRequestMessage msg, AbsolutePath outputPath, IJob job)
+
+    public ResumableDownloader(HttpRequestMessage msg, AbsolutePath outputPath, IJob job, ILogger<SingleThreadedDownloader> logger)
     {
         _job = job;
         _msg = msg;
         _outputPath = outputPath;
         _packagePath = outputPath.WithExtension(Extension.FromPath(".download_package"));
+        _logger = logger;
     }
 
     public async Task<Hash> Download(CancellationToken token)
@@ -46,10 +50,12 @@ internal class ResumableDownloader
             // Resume with different Uri in case old one is no longer valid
             downloadPackage.Address = _msg.RequestUri!.AbsoluteUri;
 
+            _logger.LogDebug("Download for {name} is resuming...", _outputPath.FileName.ToString());
             await downloader.DownloadFileTaskAsync(downloadPackage, token);
         }
         else
         {
+            _logger.LogDebug("Download for '{name}' is starting from scratch...", _outputPath.FileName.ToString());
             _outputPath.Delete();
             await downloader.DownloadFileTaskAsync(_msg.RequestUri!.AbsoluteUri, _outputPath.ToString(), token);
         }
@@ -57,13 +63,24 @@ internal class ResumableDownloader
         // Save progress if download isn't completed yet
         if (downloader.Status is DownloadStatus.Stopped or DownloadStatus.Failed)
         {
+            _logger.LogDebug("Download for '{name}' stopped before completion. Saving package...", _outputPath.FileName.ToString());
             SavePackage(downloader.Package);
-            if (_error != null && _error.GetType() != typeof(TaskCanceledException))
+            if (_error == null || _error.GetType() == typeof(TaskCanceledException))
             {
-                throw _error;
+                return new Hash();
             }
 
-            return new Hash();
+            if (_error.GetType() == typeof(NotSupportedException))
+            {
+                _logger.LogWarning("Download for '{name}' doesn't support resuming. Deleting package...", _outputPath.FileName.ToString());
+                DeletePackage();
+            }
+            else
+            {
+                _logger.LogError(_error,"Download for '{name}' encountered error. Throwing...", _outputPath.FileName.ToString());
+            }
+
+            throw _error;
         }
 
         if (downloader.Status == DownloadStatus.Completed)
@@ -76,13 +93,16 @@ internal class ResumableDownloader
             return new Hash();
         }
 
-        return await _outputPath.Open(FileMode.Open).Hash(token);
+        await using var file = _outputPath.Open(FileMode.Open);
+        return await file.Hash(token);
     }
 
     private DownloadConfiguration CreateConfiguration(HttpRequestMessage message)
     {
         var configuration = new DownloadConfiguration
         {
+            Timeout = (int)TimeSpan.FromSeconds(120).TotalMilliseconds,
+            ReserveStorageSpaceBeforeStartingDownload = true,
             RequestConfiguration = new RequestConfiguration
             {
                 Headers = message.Headers.ToWebHeaderCollection(),
@@ -109,12 +129,21 @@ internal class ResumableDownloader
         }
 
         await _job.Report(processedSize, _token);
+        if (_job.Current > _job.Size)
+        {
+            // Increase job size so progress doesn't appear stalled
+            _job.Size = (long)Math.Floor(_job.Current * 1.1);
+        }
     }
 
     private void OnDownloadStarted(object? sender, DownloadStartedEventArgs e)
     {
         _job.ResetProgress();
-        _job.Size = e.TotalBytesToReceive;
+
+        if (_job.Size < e.TotalBytesToReceive)
+        {
+            _job.Size = e.TotalBytesToReceive;
+        }
 
         // Get rid of package, since we can't use it to resume anymore
         DeletePackage();

Wabbajack.Networking.Http/SingleThreadedDownloader.cs

@@ -29,8 +29,23 @@ public class SingleThreadedDownloader : IHttpDownloader
     public async Task<Hash> Download(HttpRequestMessage message, AbsolutePath outputPath, IJob job,
         CancellationToken token)
     {
-        var downloader = new ResumableDownloader(message, outputPath, job);
+        Exception downloadError = null!;
-        return await downloader.Download(token);
+        var downloader = new ResumableDownloader(message, outputPath, job, _logger);
+        for (var i = 0; i < 3; i++)
+        {
+            try
+            {
+                return await downloader.Download(token);
+            }
+            catch (Exception ex)
+            {
+                downloadError = ex;
+                _logger.LogDebug("Download for '{name}' failed. Retrying...", outputPath.FileName.ToString());
+            }
+        }
+
+        _logger.LogError(downloadError, "Failed to download '{name}' after 3 tries.", outputPath.FileName.ToString());
+        return new Hash();
 
         // using var response = await _client.SendAsync(message, HttpCompletionOption.ResponseHeadersRead, token);
         // if (!response.IsSuccessStatusCode)

Wabbajack.Services.OSIntegrated/ResourceSettingsManager.cs

@@ -24,20 +24,20 @@ public class ResourceSettingsManager
         try
         {
             _settings ??= await _manager.Load<Dictionary<string, ResourceSetting>>("resource_settings");
-
+            if (!_settings.ContainsKey(name))
-            if (_settings.TryGetValue(name, out var found)) return found;
-
-            var newSetting = new ResourceSetting
             {
-                MaxTasks = Environment.ProcessorCount,
+                var newSetting = new ResourceSetting
-                MaxThroughput = 0
+                {
-            };
+                    MaxTasks = Environment.ProcessorCount,
+                    MaxThroughput = 0
+                };
 
-            _settings.Add(name, newSetting);
+                _settings.Add(name, newSetting);
+                await SaveSettings(_settings);
+            }
 
-            await _manager.Save("resource_settings", _settings);
+            var setting = _settings[name];
-            
+            return setting;
-            return _settings[name];
         }
         finally
         {