Merge remote-tracking branch 'upstream/master' into fix-stop-after-current-file

Duplicati/Library/Main/Operation/Backup/StreamBlock.cs

@@ -14,7 +14,7 @@
 //  You should have received a copy of the GNU Lesser General Public
 //  License along with this library; if not, write to the Free Software
 //  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-using System;
+
 using CoCoL;
 using System.IO;
 using System.Threading.Tasks;
@@ -24,9 +24,9 @@ namespace Duplicati.Library.Main.Operation.Backup
 {
     internal struct StreamProcessResult
     {
-        public string Streamhash;
-        public long Streamlength;
-        public long Blocksetid; 
+        public string Streamhash { get; internal set; }
+        public long Streamlength { get; internal set; }
+        public long Blocksetid { get; internal set; }
     }
 
     internal struct StreamBlock
@@ -40,16 +40,23 @@ namespace Duplicati.Library.Main.Operation.Backup
         public static async Task<StreamProcessResult> ProcessStream(IWriteChannel<StreamBlock> channel, string path, Stream stream, bool isMetadata, CompressionHint hint)
         {
             var tcs = new TaskCompletionSource<StreamProcessResult>();
-            await channel.WriteAsync(new StreamBlock()
+
+            // limit the stream length to that found now, a fixed point in time
+            var limitedStream = new Library.Utility.ReadLimitLengthStream(stream, stream.Length);
+            
+            var streamBlock = new StreamBlock
             {
                 Path = path,
-                Stream = stream,
+                Stream = limitedStream,
                 IsMetadata = isMetadata,
                 Hint = hint,
                 Result = tcs
-            });
-
+            };
+            
+            await channel.WriteAsync(streamBlock);
+            
             return await tcs.Task.ConfigureAwait(false);
         }
     }
+
 }

Duplicati/Library/Utility/Duplicati.Library.Utility.csproj

@@ -56,6 +56,7 @@
     <Compile Include="OverrideableStream.cs" />
     <Compile Include="ProgressReportingStream.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
+    <Compile Include="ReadLimitLengthStream.cs" />
     <Compile Include="ShaderStream.cs" />
     <Compile Include="Sizeparser.cs" />
     <Compile Include="SslCertificateValidator.cs" />

Duplicati/Library/Utility/ReadLimitLengthStream.cs

@@ -0,0 +1,151 @@
+using System;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Duplicati.Library.Utility
+{
+    // StreamReadLimitLengthWrapper() based on code from Matt Smith (Oct 26 '15 at 20:38)
+    // https://stackoverflow.com/questions/33354822/how-to-set-length-in-stream-without-truncated
+
+    public class ReadLimitLengthStream : Stream
+    {
+        readonly Stream m_innerStream;
+        readonly long m_endPosition;
+
+        public ReadLimitLengthStream(Stream innerStream, long size)
+        {
+            if (size < 0)
+            {
+                throw new ArgumentOutOfRangeException(nameof(size));
+            }
+
+            m_innerStream = innerStream ?? throw new ArgumentNullException(nameof(innerStream));
+            m_endPosition = m_innerStream.Position + size;
+        }
+
+        public override bool CanRead => m_innerStream.CanRead;
+
+        public override bool CanSeek => m_innerStream.CanSeek;
+
+        public override bool CanWrite => false;
+
+        public override void Flush()
+        {
+            // NOOP
+        }
+
+        public override long Length => m_endPosition;
+
+        public override long Position
+        {
+            get => m_innerStream.Position;
+            set => m_innerStream.Position = value;
+        }
+
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            count = GetAllowedCount(count);
+            return m_innerStream.Read(buffer, offset, count);
+        }
+
+        public override long Seek(long offset, SeekOrigin origin)
+        {
+            return m_innerStream.Seek(offset, origin);
+        }
+
+        public override void SetLength(long value)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override bool CanTimeout => m_innerStream.CanTimeout;
+
+        public override int ReadTimeout
+        {
+            get => m_innerStream.ReadTimeout;
+            set => m_innerStream.ReadTimeout = value;
+        }
+
+        public override int WriteTimeout
+        {
+            get => m_innerStream.ReadTimeout;
+            set => m_innerStream.ReadTimeout = value;
+        }
+
+        public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
+        {
+            count = GetAllowedCount(count);
+            return m_innerStream.BeginRead(buffer, offset, count, callback, state);
+        }
+
+        public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback callback, object state)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override void Close()
+        {
+            // Since this wrapper does not own the underlying stream, we do not want it to close the underlying stream
+        }
+
+        public override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken)
+        {
+            return m_innerStream.CopyToAsync(destination, bufferSize, cancellationToken);
+        }
+
+        public override int EndRead(IAsyncResult asyncResult)
+        {
+            return m_innerStream.EndRead(asyncResult);
+        }
+
+        public override Task FlushAsync(CancellationToken cancellationToken)
+        {
+            return m_innerStream.FlushAsync(cancellationToken);
+        }
+
+        public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            count = GetAllowedCount(count);
+            return m_innerStream.ReadAsync(buffer, offset, count, cancellationToken);
+        }
+
+        public override int ReadByte()
+        {
+            var count = GetAllowedCount(1);
+            if (count == 0)
+            {
+                return -1;
+            }
+
+            return m_innerStream.ReadByte();
+        }
+
+        public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        {
+            throw new NotSupportedException();
+        }
+
+        public override void WriteByte(byte value)
+        {
+            throw new NotSupportedException();
+        }
+
+        private int GetAllowedCount(int count)
+        {
+            long pos = m_innerStream.Position;
+            long maxCount = m_endPosition - pos;
+            if (count > maxCount)
+            {
+                return (int)maxCount;
+            }
+
+            return count;
+        }
+    }
+}

Duplicati/UnitTest/TestUtils.cs

@@ -21,6 +21,8 @@ using Duplicati.Library.Utility;
 using System.Linq;
 using Duplicati.Library.Logging;
 using System.Reflection;
+using System.Threading;
+using System.Threading.Tasks;
 using Duplicati.Library.Common.IO;
 
 namespace Duplicati.UnitTest
@@ -46,6 +48,30 @@ namespace Duplicati.UnitTest
             }
         }
 
+        public static async Task GrowingFile(string testFile, CancellationToken token)
+        {
+            try
+            {
+                var str = new string('*', 50);
+                while (true)
+                {
+                    if (token.IsCancellationRequested)
+                    {
+                        continue;
+                    }
+                    File.AppendAllText(testFile, str);
+                    await Task.Delay(18, token).ConfigureAwait(false);
+                }
+            }
+            finally
+            {
+                if (File.Exists(testFile))
+                {
+                    File.Delete(testFile);
+                }
+            }
+        }
+
         public static string GetDefaultTarget(string other = null)
         {
             string alttarget = System.IO.Path.Combine(Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location), "unittest_target.txt");

Duplicati/UnitTest/UtilityTests.cs

@@ -22,6 +22,8 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Globalization;
 using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace Duplicati.UnitTest
 {
@@ -94,6 +96,35 @@ namespace Duplicati.UnitTest
             }
         }
 
+        [Test]
+        [Category("Utility")]
+        public static void ReadLimitLengthStreamWithGrowingFile()
+        {
+            var growingFilename = Path.GetTempFileName();
+            long fixedGrowingStreamLength, limitStreamLength, nextGrowingStreamLength;
+            using (CancellationTokenSource tokenSource = new CancellationTokenSource())
+            {
+                Task task = TestUtils.GrowingFile(growingFilename, tokenSource.Token);
+                Thread.Sleep(100);
+                using (FileStream growingStream = new FileStream(growingFilename, FileMode.Open, FileAccess.Read, FileShare.ReadWrite))
+                {
+                    Thread.Sleep(100);
+                    fixedGrowingStreamLength = growingStream.Length;
+                    Thread.Sleep(100);
+                    using (var limitStream = new ReadLimitLengthStream(growingStream, fixedGrowingStreamLength))
+                    {
+                        Thread.Sleep(100);
+                        nextGrowingStreamLength = growingStream.Length;
+                        limitStreamLength = limitStream.Length;
+                    }
+                }
+            }
+            Console.WriteLine($"fixedGrowingStreamLength:{fixedGrowingStreamLength} limitStreamLength:{limitStreamLength} nextGrowingStreamLength:{nextGrowingStreamLength}");
+            Assert.IsTrue(fixedGrowingStreamLength > 0);
+            Assert.AreEqual(fixedGrowingStreamLength, limitStreamLength);
+            Assert.IsTrue(limitStreamLength < nextGrowingStreamLength);
+        }
+
         [Test]
         [Category("Utility")]
         public static void ForceStreamRead()