[Mocha] Fix RabbitMQ recovery, publishing and outbox dispatch

src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/MessageBusOutboxWorker.cs

@@ -12,6 +12,7 @@ internal sealed class PostgresMessageBusOutboxWorker(PostgresOutboxProcessor pro
 {
     private readonly object _lock = new();
     private ContinuousTask? _task;
+    private Task _processing = Task.CompletedTask;
 
     /// <summary>
     /// Starts the outbox processing background task. This call is idempotent: invoking it again
@@ -28,14 +29,15 @@ public Task StartAsync(CancellationToken cancellationToken)
                 return Task.CompletedTask;
             }
 
-            _task = new ContinuousTask(processor.ProcessAsync);
+            _task = new ContinuousTask(ProcessAsync);
         }
 
         return Task.CompletedTask;
     }
 
     /// <summary>
-    /// Stops the outbox processing background task and waits for it to complete gracefully.
+    /// Stops the outbox processing background task and waits until the messages that were being dispatched
+    /// are either removed from the outbox or released.
     /// </summary>
     /// <param name="cancellationToken">A token that signals when shutdown should be forced.</param>
     public async Task StopAsync(CancellationToken cancellationToken)
@@ -54,5 +56,18 @@ public async Task StopAsync(CancellationToken cancellationToken)
         }
 
         await task.DisposeAsync();
+
+        // The continuous task stops waiting once it is cancelled, while the processor still removes the
+        // messages a transport has already accepted. The wait ends early when the shutdown is forced.
+        await Volatile.Read(ref _processing)
+            .WaitAsync(cancellationToken)
+            .ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
+    }
+
+    private Task ProcessAsync(CancellationToken cancellationToken)
+    {
+        var processing = processor.ProcessAsync(cancellationToken);
+        Volatile.Write(ref _processing, processing);
+        return processing;
     }
 }