Merge pull request #6165 from duplicati/feature/update-minio-6.0.4

Updated Minio to 6.0.4

Duplicati/Library/Backend/S3/Duplicati.Library.Backend.S3.csproj

@@ -13,7 +13,7 @@
     <PackageReference Include="AWSSDK.Core" Version="3.7.400.45" />
     <PackageReference Include="AWSSDK.IdentityManagement" Version="3.7.402.39" />
     <PackageReference Include="AWSSDK.S3" Version="3.7.405.9" />
-    <PackageReference Include="Minio" Version="6.0.2" />
+    <PackageReference Include="Minio" Version="6.0.4" />
   </ItemGroup>
   
   <ItemGroup>

Duplicati/Library/Backend/S3/S3MinioClient.cs

@@ -19,7 +19,6 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER 
 // DEALINGS IN THE SOFTWARE.
 using System.Runtime.CompilerServices;
-using System.Threading.Channels;
 using Duplicati.Library.Common.IO;
 using Duplicati.Library.Interface;
 using Duplicati.Library.Utility;
@@ -56,27 +55,6 @@ namespace Duplicati.Library.Backend
             m_dnsHost = servername;
         }
 
-        private static async IAsyncEnumerable<T> ToAsyncEnumerable<T>(
-            IObservable<T> observable,
-            [EnumeratorCancellation] CancellationToken cancellationToken = default)
-        {
-            var channel = Channel.CreateUnbounded<T>(); // Buffered channel for async iteration
-
-            using var subscription = observable.Subscribe(
-                item => channel.Writer.TryWrite(item),
-                ex => channel.Writer.TryComplete(ex),
-                () => channel.Writer.TryComplete()
-            );
-
-            while (await channel.Reader.WaitToReadAsync(cancellationToken))
-            {
-                while (channel.Reader.TryRead(out var item))
-                {
-                    yield return item;
-                }
-            }
-        }
-
         public async IAsyncEnumerable<IFileEntry> ListBucketAsync(string bucketName, string prefix, bool recursive, [EnumeratorCancellation] CancellationToken cancellationToken)
         {
             await ThrowExceptionIfBucketDoesNotExist(bucketName, cancellationToken).ConfigureAwait(false);
@@ -84,11 +62,7 @@ namespace Duplicati.Library.Backend
             if (!string.IsNullOrWhiteSpace(prefix))
                 prefix = Util.AppendDirSeparator(prefix, "/");
 
-            var observable = await Utility.Utility.WithTimeout(m_timeouts.ListTimeout, cancellationToken, ct
-                => m_client.ListObjectsAsync(new ListObjectsArgs().WithBucket(bucketName).WithPrefix(prefix).WithRecursive(recursive), ct)
-            ).ConfigureAwait(false);
-
-            await foreach (var obj in ToAsyncEnumerable(observable, cancellationToken).ConfigureAwait(false))
+            await foreach (var obj in Utility.Utility.WithPerItemTimeout(m_client.ListObjectsEnumAsync(new ListObjectsArgs().WithBucket(bucketName).WithPrefix(prefix).WithRecursive(recursive), cancellationToken), m_timeouts.ListTimeout, cancellationToken))
             {
                 if (obj.Key == prefix || !obj.Key.StartsWith(prefix))
                     continue;

Duplicati/Library/Utility/Utility.cs

@@ -29,6 +29,7 @@ using System.Linq;
 using System.Net;
 using System.Net.NetworkInformation;
 using System.Net.Sockets;
+using System.Runtime.CompilerServices;
 using System.Runtime.Versioning;
 using System.Security.Cryptography;
 using System.Security.Cryptography.X509Certificates;
@@ -1766,6 +1767,47 @@ namespace Duplicati.Library.Utility
             }
         }
 
+        /// <summary>
+        /// Wraps an async enumerable in a timeout observing enumerable
+        /// </summary>
+        /// <typeparam name="T">The type of the items in the enumerable</typeparam>
+        /// <param name="source">The source enumerable</param>
+        /// <param name="timeoutPerItem">The timeout to observe for each item</param>
+        /// <param name="outerToken">The cancellation token for the outer operation</param>
+        /// <returns>The wrapped enumerable</returns>
+        public static async IAsyncEnumerable<T> WithPerItemTimeout<T>(
+            IAsyncEnumerable<T> source,
+            TimeSpan timeoutPerItem,
+            [EnumeratorCancellation] CancellationToken outerToken)
+        {
+            using var timeoutPolicy = new CancellationTokenSource();
+            using var linked = CancellationTokenSource.CreateLinkedTokenSource(timeoutPolicy.Token, outerToken);
+
+            await using var enumerator = await WithTimeout(timeoutPerItem, outerToken, _ => source.GetAsyncEnumerator(linked.Token)).ConfigureAwait(false);
+            while (true)
+            {
+                timeoutPolicy.CancelAfter(timeoutPerItem);
+                Task<bool> moveNextTask;
+                try
+                {
+                    moveNextTask = enumerator.MoveNextAsync().AsTask();
+                    var completed = await Task.WhenAny(moveNextTask, Task.Delay(Timeout.Infinite, timeoutPolicy.Token));
+                    if (completed != moveNextTask || timeoutPolicy.IsCancellationRequested)
+                        throw new TimeoutException($"Timeout while waiting for next item ({timeoutPerItem.TotalSeconds}s)");
+
+                    if (!moveNextTask.Result)
+                        break;
+
+                    yield return enumerator.Current;
+                }
+                finally
+                {
+                    timeoutPolicy.Token.ThrowIfCancellationRequested();
+                    timeoutPolicy.CancelAfter(Timeout.Infinite); // Reset for next item
+                }
+            }
+        }
+
         /// <summary>
         /// Wraps the stream in a timeout observing stream
         /// </summary>