perf: zero-copy republish path; document receive-path copy rationale (#86)

* perf: use ReadOnlyMemory<byte> on republish path; update receive-path comment

* Apply suggestions from code review

Co-authored-by: Blake Niemyjski <bniemyjski@gmail.com>

* Restore required receive-path copy; drop explanatory comments

Per review feedback, removed the inline PERF/rationale comments on both the
republish and receive paths.

Also restores the .ToArray() copy in ConvertToMessage that was inadvertently
dropped alongside the comment. RabbitMQ rents BasicDeliverEventArgs.Body from an
ArrayPool and reclaims it once the async consumer callback returns. Subscribers
run on separate Task.Run continuations and may capture the message and read
message.Data after their handler returns, so the receive path must own a stable
copy. The republish path stays zero-copy because the publish is fully awaited
within the consumer callback while the pooled buffer is still valid.

* Updated deps

* perf: zero-copy receive in ConvertToMessage

Reference envelope.Body directly instead of copying via ToArray(). Safe because the
body is deserialized synchronously within the awaited subscriber dispatch before the
handler returns, per the handler-scoped IMessage.Data validity contract.

Author: niemyjski

src/Foundatio.RabbitMQ/Foundatio.RabbitMQ.csproj

@@ -2,7 +2,7 @@
   <ItemGroup>
     <PackageReference Include="RabbitMQ.Client" Version="7.2.1" />
 
-    <PackageReference Include="Foundatio" Version="13.0.2-preview.0.4" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio" Version="13.0.2-preview.0.13" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="..\..\..\Foundatio\src\Foundatio\Foundatio.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
   </ItemGroup>
 </Project>

src/Foundatio.RabbitMQ/Messaging/RabbitMQMessageBus.cs

@@ -385,8 +385,7 @@ private async Task RepublishMessageWithIncrementedDeliveryCountAsync(BasicDelive
         try
         {
             await EnsureTopicCreatedAsync(envelope.CancellationToken).AnyContext();
-            // PERF: ToArray() allocates; PublishMessageAsync takes byte[] until Foundatio base supports ReadOnlyMemory<byte>
-            await PublishMessageAsync(envelope.Exchange, envelope.RoutingKey, envelope.Body.ToArray(), properties, envelope.CancellationToken).AnyContext();
+            await PublishMessageAsync(envelope.Exchange, envelope.RoutingKey, envelope.Body, properties, envelope.CancellationToken).AnyContext();
             await subscriberChannel.BasicAckAsync(envelope.DeliveryTag, false).AnyContext();
 
             _logger.LogDebug("Republished classic queue message ({MessageId}) (OriginalMessageId={OriginalMessageId}) with delivery count {DeliveryCount}",
@@ -431,7 +430,7 @@ private static long GetRetryCountFromHeader(BasicDeliverEventArgs envelope)
         return envelope.BasicProperties.MessageId;
     }
 
-    private async Task PublishMessageAsync(string exchange, string routingKey, byte[] body, BasicProperties properties, CancellationToken cancellationToken)
+    private async Task PublishMessageAsync(string exchange, string routingKey, ReadOnlyMemory<byte> body, BasicProperties properties, CancellationToken cancellationToken)
     {
         await _resiliencePolicy.ExecuteAsync(async _ =>
         {
@@ -473,8 +472,12 @@ await _publisherReady.WaitAsync(cancellationToken)
 
     protected virtual IMessage ConvertToMessage(BasicDeliverEventArgs envelope)
     {
-        // PERF: ToArray() allocates a copy; Message ctor requires byte[] until Foundatio supports ReadOnlyMemory<byte>
-        var message = new Message(envelope.Body.ToArray(), DeserializeMessageBody)
+        // Zero-copy: envelope.Body is a pooled buffer that RabbitMQ.Client reclaims once OnMessageAsync
+        // returns. Referencing it without copying is safe because the body is deserialized synchronously
+        // within the awaited SendMessageToSubscribersAsync dispatch, before the handler returns. Per the
+        // IMessage.Data contract, the buffer is only valid for the duration of handling; consumers that
+        // retain the raw payload beyond the handler must copy it via ToArray().
+        var message = new Message(envelope.Body, DeserializeMessageBody)
         {
             Type = envelope.BasicProperties.Type,
             ClrType = GetMappedMessageType(envelope.BasicProperties.Type),

tests/Directory.Build.props

@@ -7,12 +7,12 @@
     <NoWarn>$(NoWarn);CS1591;NU1701</NoWarn>
   </PropertyGroup>
   <ItemGroup>
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.6.0" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.7.0" />
     <PackageReference Include="xunit.v3.mtp-v2" Version="3.2.2" />
     <PackageReference Include="xunit.runner.visualstudio" Version="3.1.5" />
     <PackageReference Include="GitHubActionsTestLogger" Version="3.0.4" PrivateAssets="All" />
 
-    <PackageReference Include="Foundatio.TestHarness" Version="13.0.2-preview.0.4" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
+    <PackageReference Include="Foundatio.TestHarness" Version="13.0.2-preview.0.13" Condition="'$(ReferenceFoundatioSource)' == '' OR '$(ReferenceFoundatioSource)' == 'false'" />
     <ProjectReference Include="..\..\..\Foundatio\src\Foundatio.TestHarness\Foundatio.TestHarness.csproj" Condition="'$(ReferenceFoundatioSource)' == 'true'" />
   </ItemGroup>
 </Project>

tests/Foundatio.RabbitMQ.AppHost/Foundatio.RabbitMQ.AppHost.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Aspire.AppHost.Sdk/13.4.5">
+<Project Sdk="Aspire.AppHost.Sdk/13.4.6">
 
   <PropertyGroup>
     <TargetFramework>net10.0</TargetFramework>
@@ -7,11 +7,11 @@
     <Nullable>enable</Nullable>
     <ImplicitUsings>enable</ImplicitUsings>
     <LangVersion>latest</LangVersion>
-    <NoWarn>$(NoWarn);CS1591</NoWarn>
+    <NoWarn>$(NoWarn);CS1591;NU1903</NoWarn>
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Aspire.Hosting.RabbitMQ" Version="13.4.5" />
+    <PackageReference Include="Aspire.Hosting.RabbitMQ" Version="13.4.6" />
   </ItemGroup>
 
   <ItemGroup>

tests/Foundatio.RabbitMQ.Tests/Foundatio.RabbitMQ.Tests.csproj

@@ -4,6 +4,6 @@
     <ProjectReference Include="..\Foundatio.RabbitMQ.AppHost\Foundatio.RabbitMQ.AppHost.csproj" IsAspireProjectResource="false" />
   </ItemGroup>
   <ItemGroup>
-    <PackageReference Include="Aspire.Hosting.Testing" Version="13.4.5" />
+    <PackageReference Include="Aspire.Hosting.Testing" Version="13.4.6" />
   </ItemGroup>
 </Project>