Add guard rails for DataStreams that differ in-size from their reported size.

source/Halibut.Tests/DataStreamFixture.cs

@@ -1,7 +1,10 @@
-using System;
+using System;
+using System.Diagnostics;
 using System.IO;
+using System.Linq;
 using System.Threading.Tasks;
 using FluentAssertions;
+using Halibut.Diagnostics;
 using Halibut.Tests.Support;
 using Halibut.Tests.Support.TestAttributes;
 using Halibut.Tests.Support.TestCases;
@@ -50,5 +53,198 @@ public async Task ASyncDataStreamWriter_CanBeUsedInAsync()
             await ((IDataStreamInternal) ds).TransmitAsync(memoryStream, CancellationToken);
             memoryStream.ToArray().Should().BeEquivalentTo(data);
         }
+
+        [Test]
+        [LatestClientAndLatestServiceTestCases(testNetworkConditions: false)]
+        public async Task WhenSendingADataStream_AndWeSendMoreDataThanWeShould_ThenADescriptiveErrorIsLogged(ClientAndServiceTestCase clientAndServiceTestCase)
+        {
+            await using (var clientAndService = await clientAndServiceTestCase.CreateTestCaseBuilder()
+                       .WithStandardServices()
+                       .AsLatestClientAndLatestServiceBuilder()
+                       .WithHalibutTimeoutsAndLimits(new HalibutTimeoutsAndLimitsForTestsBuilder()
+                           .Build()
+                           .Apply(h => h.ThrowOnDataStreamSizeMismatch = false))
+                       .RecordingClientLogs(out var clientLogs)
+                       .RecordingServiceLogs(out var serviceLogs)
+                       .Build(CancellationToken))
+            {
+                var readDataStreamService = clientAndService.CreateAsyncClient<IReadDataStreamService, IAsyncClientReadDataStreamService>();
+
+                var actualData = new byte[100];
+                new Random().NextBytes(actualData);
+
+                var maliciousDataStream = new DataStream(10, async (stream, ct) =>
+                {
+                    await stream.WriteAsync(actualData, 0, actualData.Length, ct);
+                });
+
+                await AssertException.Throws<HalibutClientException>(async () => 
+                    await readDataStreamService.SendDataAsync(maliciousDataStream));
+
+                var allClientLogs = clientLogs.Values.SelectMany(log => log.GetLogs()).ToList();
+                allClientLogs.Should().Contain(log => 
+                    log.Type == EventType.Error && 
+                    log.FormattedMessage.Contains("Data stream size mismatch detected during send") &&
+                    log.FormattedMessage.Contains("Declared length: 10") &&
+                    log.FormattedMessage.Contains("Actual bytes written: 100"));
+
+                var allServiceLogs = serviceLogs.Values.SelectMany(log => log.GetLogs()).ToList();
+                allServiceLogs.Should().Contain(log => 
+                    log.Type == EventType.Error && 
+                    log.FormattedMessage.Contains("Data stream size mismatch detected") &&
+                    log.FormattedMessage.Contains("Message ID:") &&
+                    log.FormattedMessage.Contains("Stream ID:") &&
+                    log.FormattedMessage.Contains("Expected length: 10") &&
+                    log.FormattedMessage.Contains("Total length of all DataStreams"));
+
+            }
+        }
+
+        [Test]
+        [LatestClientAndLatestServiceTestCases(testNetworkConditions: false)]
+        public async Task WhenSendingADataStream_AndWeSendLessDataThanWeShould_ThenADescriptiveErrorIsLogged(ClientAndServiceTestCase clientAndServiceTestCase)
+        {
+            await using (var clientAndService = await clientAndServiceTestCase.CreateTestCaseBuilder()
+                       .WithStandardServices()
+                       .AsLatestClientAndLatestServiceBuilder()
+                       .WithHalibutTimeoutsAndLimits(new HalibutTimeoutsAndLimitsForTestsBuilder()
+                           .Build()
+                           .Apply(h => h.ThrowOnDataStreamSizeMismatch = false))
+                       .RecordingClientLogs(out var clientLogs)
+                       .RecordingServiceLogs(out var serviceLogs)
+                       .Build(CancellationToken))
+            {
+                var readDataStreamService = clientAndService.CreateAsyncClient<IReadDataStreamService, IAsyncClientReadDataStreamService>();
+
+                var actualData = new byte[10];
+                new Random().NextBytes(actualData);
+
+                var underSizedDataStream = new DataStream(100, async (stream, ct) =>
+                {
+                    await stream.WriteAsync(actualData, 0, actualData.Length, ct);
+                });
+
+                await AssertException.Throws<HalibutClientException>(async () => 
+                    await readDataStreamService.SendDataAsync(underSizedDataStream));
+
+                var allClientLogs = clientLogs.Values.SelectMany(log => log.GetLogs()).ToList();
+                allClientLogs.Should().Contain(log => 
+                    log.Type == EventType.Error && 
+                    log.FormattedMessage.Contains("Data stream size mismatch detected during send") &&
+                    log.FormattedMessage.Contains("Declared length: 100") &&
+                    log.FormattedMessage.Contains("Actual bytes written: 10"));
+            }
+        }
+
+        [Test]
+        [LatestClientAndLatestServiceTestCases(testNetworkConditions: false)]
+        public async Task WhenSendingADataStream_AndWeSendMoreDataThanWeShould_AndThrowOnDataStreamSizeMismatchIsEnabled_ThenItThrows(ClientAndServiceTestCase clientAndServiceTestCase)
+        {
+            await using (var clientAndService = await clientAndServiceTestCase.CreateTestCaseBuilder()
+                       .WithStandardServices()
+                       .AsLatestClientAndLatestServiceBuilder()
+                       .WithHalibutTimeoutsAndLimits(new HalibutTimeoutsAndLimits
+                       {
+                           ThrowOnDataStreamSizeMismatch = true
+                       })
+                       .RecordingClientLogs(out var clientLogs)
+                       .Build(CancellationToken))
+            {
+                var readDataStreamService = clientAndService.CreateAsyncClient<IReadDataStreamService, IAsyncClientReadDataStreamService>();
+
+                var actualData = new byte[100];
+                new Random().NextBytes(actualData);
+
+                var maliciousDataStream = new DataStream(10, async (stream, ct) =>
+                {
+                    await stream.WriteAsync(actualData, 0, actualData.Length, ct);
+                });
+
+                var exception = await AssertException.Throws<HalibutClientException>(async () => 
+                    await readDataStreamService.SendDataAsync(maliciousDataStream));
+
+                exception.And.Message.Should().Contain("Data stream size mismatch");
+            }
+        }
+
+        [Test]
+        [LatestClientAndLatestServiceTestCases(testNetworkConditions: false)]
+        public async Task WhenSendingADataStream_AndWeSendLessDataThanWeShould_AndThrowOnDataStreamSizeMismatchIsEnabled_ThenItThrows(ClientAndServiceTestCase clientAndServiceTestCase)
+        {
+            await using (var clientAndService = await clientAndServiceTestCase.CreateTestCaseBuilder()
+                       .WithStandardServices()
+                       .AsLatestClientAndLatestServiceBuilder()
+                       .WithHalibutTimeoutsAndLimits(new HalibutTimeoutsAndLimits
+                       {
+                           ThrowOnDataStreamSizeMismatch = true
+                       })
+                       .RecordingClientLogs(out var clientLogs)
+                       .Build(CancellationToken))
+            {
+                var readDataStreamService = clientAndService.CreateAsyncClient<IReadDataStreamService, IAsyncClientReadDataStreamService>();
+
+                var actualData = new byte[10];
+                new Random().NextBytes(actualData);
+
+                var underSizedDataStream = new DataStream(100, async (stream, ct) =>
+                {
+                    await stream.WriteAsync(actualData, 0, actualData.Length, ct);
+                });
+
+                var exception = await AssertException.Throws<HalibutClientException>(async () => 
+                    await readDataStreamService.SendDataAsync(underSizedDataStream));
+
+                exception.And.Message.Should().Contain("Data stream size mismatch");
+            }
+        }
+
+        [Test]
+        [LatestClientAndLatestServiceTestCases(testNetworkConditions: false)]
+        public async Task WhenSendingADataStream_AndWeSendLessDataThanWeShould_AndThrowOnDataStreamSizeMismatchIsEnabled_ThenSecondRequestSucceedsQuickly(ClientAndServiceTestCase clientAndServiceTestCase)
+        {
+            await using (var clientAndService = await clientAndServiceTestCase.CreateTestCaseBuilder()
+                       .WithStandardServices()
+                       .AsLatestClientAndLatestServiceBuilder()
+                       .WithHalibutTimeoutsAndLimits(new HalibutTimeoutsAndLimits
+                       {
+                           ThrowOnDataStreamSizeMismatch = true,
+                           TcpClientReceiveResponseTimeout = TimeSpan.FromSeconds(60)
+                       })
+                       .RecordingClientLogs(out var clientLogs)
+                       .Build(CancellationToken))
+            {
+                var readDataStreamService = clientAndService.CreateAsyncClient<IReadDataStreamService, IAsyncClientReadDataStreamService>();
+
+                var underSizedData = new byte[10];
+                new Random().NextBytes(underSizedData);
+
+                var underSizedDataStream = new DataStream(100, async (stream, ct) =>
+                {
+                    await stream.WriteAsync(underSizedData, 0, underSizedData.Length, ct);
+                });
+
+                var stopwatch = Stopwatch.StartNew();
+
+                await AssertException.Throws<HalibutClientException>(async () => 
+                    await readDataStreamService.SendDataAsync(underSizedDataStream));
+
+                var correctData = new byte[50];
+                new Random().NextBytes(correctData);
+                var correctDataStream = new DataStream(50, async (stream, ct) =>
+                {
+                    await stream.WriteAsync(correctData, 0, correctData.Length, ct);
+                });
+
+                var received = await readDataStreamService.SendDataAsync(correctDataStream);
+
+                var secondRequestTime = stopwatch.Elapsed;
+                received.Should().Be(50);
+                Logger.Information("All requests took: {Time}s", secondRequestTime);
+                secondRequestTime.Should().BeLessThan(TimeSpan.FromSeconds(30), 
+                    "We should detect the wrong size DataStream, this results in an immediate exception being raised." +
+                    " This will result in the client killing the connection, which will result in the client also receiving a EOF. " +
+                    "This will result in both sides quickly re-connecting allowing another RPC to be re-attempted quickly.");
+            }
+        }
     }
 }

source/Halibut.Tests/Diagnostics/ExceptionReturnedByHalibutProxyExtensionMethodFixture.cs

@@ -122,7 +122,8 @@ public async Task WhenTheConnectionPausesWaitingForAResponse(ClientAndServiceTes
                         .Be(HalibutNetworkExceptionType.IsNetworkError);
 
                     exception.Message.Should().ContainAny(
-                        "Unable to read data from the transport connection: Connection timed out.", 
+                        "Unable to read data from the transport connection: Connection timed out.",
+                        "Unable to read data from the transport connection: Operation timed out.",
                         "Unable to read data from the transport connection: A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond.");
                 }
             }

source/Halibut.Tests/ExceptionContractFixture.cs

@@ -60,6 +60,7 @@ public async Task WhenThePollingRequestHasBegunTransfer_AndATcpTimeoutIsReached_
             (await AssertException.Throws<HalibutClientException>(async () => await doSomeActionClient.ActionAsync(new(CancellationToken))))
                 .And.Message.Should().ContainAny(
                     "Unable to read data from the transport connection: Connection timed out.",
+                    "Unable to read data from the transport connection: Operation timed out.",
                     "Unable to read data from the transport connection: A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond");
 
             waitSemaphore.Release();

source/Halibut.Tests/FailureModesFixture.cs

@@ -131,7 +131,11 @@ public async Task FailOnInvalidHostname()
                 else
                 {
                     // Failed with: An error occurred when sending a request to 'https://sduj08ud9382ujd98dw9fh934hdj2389u982:8000/', before the request could begin: Name or service not known, but found False.
-                    new [] {"No such device or address", "Resource temporarily unavailable", "Name or service not known"}.Any(message.Contains).Should().BeTrue($"Message does not match known strings: {message}");
+                    new [] {"No such device or address", 
+                        "Resource temporarily unavailable", 
+                        "Name or service not known",
+                        "nodename nor servname provided, or not known"
+                    }.Any(message.Contains).Should().BeTrue($"Message does not match known strings: {message}");
                 }
             }
         }

source/Halibut.Tests/PollingServiceTimeoutsFixture.cs

@@ -120,6 +120,7 @@ public async Task WhenThePollingRequestHasBegunTransfer_AndWeTimeoutWaitingForTh
                 var stopwatch = Stopwatch.StartNew();
                 var exception = (await AssertException.Throws<HalibutClientException>(async () => await doSomeActionClient.ActionAsync(new(CancellationToken)))).And;
                 exception.Message.Should().ContainAny(
+                        "Unable to read data from the transport connection: Operation timed out.",
                         "Unable to read data from the transport connection: Connection timed out.",
                         "Unable to read data from the transport connection: A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond");
                 exception.ConnectionState.Should().Be(ConnectionState.Unknown);

source/Halibut.Tests/ProxyFixture.cs

@@ -85,6 +85,7 @@ public async Task ClientTimesOutConnectingToAProxy_WhenTheProxyHangsDuringConnec
                         "No connection could be made because the target machine actively refused it",
                         "the polling endpoint did not collect the request within the allowed time",
                         "Unable to read data from the transport connection: Connection timed out.",
+                        "Unable to read data from the transport connection: Operation timed out.",
                         "A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond.",
                         "A timeout while waiting for the proxy server at");
                 ;

source/Halibut.Tests/Timeouts/ReceiveResponseTimeoutTests.cs

@@ -39,6 +39,7 @@ public async Task WhenRpcExecutionExceedsReceiveResponseTimeout_ThenInitialDataR
                 (await AssertionExtensions.Should(() => doSomeActionClient.ActionAsync()).ThrowAsync<HalibutClientException>())
                     .And.Message.Should().ContainAny(
                         "Connection timed out.",
+                        "Operation timed out.",
                         "A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond");
             }
         }
@@ -85,6 +86,7 @@ public async Task WhenRpcExecutionIsWithinReceiveResponseTimeout_ButSubsequentDa
                 (await AssertionExtensions.Should(() => lastServiceClient.GetListAsync()).ThrowAsync<HalibutClientException>())
                     .And.Message.Should().ContainAny(
                         "Connection timed out.",
+                        "Operation timed out.",
                         "A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond",
                         "Unable to read data from the transport connection: Connection reset by peer");
             }
@@ -117,6 +119,7 @@ public async Task WhenRpcExecutionIsWithinReceiveResponseTimeout_ButDataStreamDa
                 Logger.Information(e, "The received expected exception, we were expecting one");
                 e.Message.Should().ContainAny(
                         "Connection timed out.",
+                        "Operation timed out.",
                         "A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond");
             }
         }

source/Halibut.Tests/Timeouts/SendingAndReceivingRequestMessagesTimeoutsFixture.cs

@@ -202,6 +202,7 @@ static void AssertExceptionLooksLikeAWriteTimeout(HalibutClientException? e)
         {
             e!.Message.Should().ContainAny(
                 "Unable to write data to the transport connection: Connection timed out.",
+                "Unable to write data to the transport connection: Operation timed out.",
                 " Unable to write data to the transport connection: A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond");
 
             e.IsNetworkError().Should().Be(HalibutNetworkExceptionType.IsNetworkError);
@@ -210,6 +211,7 @@ static void AssertExceptionLooksLikeAWriteTimeout(HalibutClientException? e)
         static void AssertExceptionMessageLooksLikeAReadTimeout(HalibutClientException? e)
         {
             e!.Message.Should().ContainAny(
+                "Unable to read data from the transport connection: Operation timed out.",
                 "Unable to read data from the transport connection: Connection timed out.",
                 "Unable to read data from the transport connection: A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond.");
 

source/Halibut.Tests/Transport/Protocol/MessageSerializerFixture.cs

@@ -104,7 +104,8 @@ public async Task BackwardsCompatibility_ExtraParametersInServerErrorAreIgnored(
             await using (var stream = new RewindableBufferStream(new MemoryStream(Convert.FromBase64String(base64Bson))))
             {
                 var result = await ReadMessage<ResponseMessage>(sut, stream);
-                result.Error.Should().NotBeNull();
+                result.Should().NotBeNull();
+                result!.Error.Should().NotBeNull();
                 result.Error!.Message = "foo";
                 result.Error.HalibutErrorType = "MethodNotFoundHalibutClientException";
             }
@@ -303,7 +304,7 @@ static byte[] DeflateString(string s)
             }
         }
 
-        async Task<T> ReadMessage<T>(MessageSerializer messageSerializer, RewindableBufferStream rewindableBufferStream)
+        async Task<T?> ReadMessage<T>(MessageSerializer messageSerializer, RewindableBufferStream rewindableBufferStream)
         {
             return (await messageSerializer.ReadMessageAsync<T>(rewindableBufferStream, CancellationToken)).Message;
         }