[Messenger] Fix NoAutoAckStamp handling in Worker::flush()

The line `$envelope = $envelope->withoutAll(NoAutoAckStamp::class);` was modifying a local variable that was never used after assignment. This change moves the stamp removal to the catch block where it's actually needed, ensuring envelopes stored for acknowledgment don't contain contradictory NoAutoAckStamp.

Author: wazum

src/Symfony/Component/Messenger/Tests/Fixtures/DummyReceiver.php

@@ -63,4 +63,9 @@ public function getAcknowledgedEnvelopes(): array
     {
         return $this->acknowledgedEnvelopes;
     }
+
+    public function getRejectedEnvelopes(): array
+    {
+        return $this->rejectedEnvelopes;
+    }
 }

src/Symfony/Component/Messenger/Tests/WorkerTest.php

@@ -38,6 +38,8 @@
 use Symfony\Component\Messenger\MessageBusInterface;
 use Symfony\Component\Messenger\Middleware\HandleMessageMiddleware;
 use Symfony\Component\Messenger\Stamp\ConsumedByWorkerStamp;
+use Symfony\Component\Messenger\Stamp\FlushBatchHandlersStamp;
+use Symfony\Component\Messenger\Stamp\NoAutoAckStamp;
 use Symfony\Component\Messenger\Stamp\ReceivedStamp;
 use Symfony\Component\Messenger\Stamp\SentStamp;
 use Symfony\Component\Messenger\Stamp\StampInterface;
@@ -584,6 +586,49 @@ public function testFlushBatchOnStop()
 
         $this->assertSame($expectedMessages, $handler->processedMessages);
     }
+
+    public function testFlushRemovesNoAutoAckStampOnException()
+    {
+        $envelope = new Envelope(new DummyMessage('Test'));
+        $receiver = new DummyReceiver([[$envelope]]);
+
+        $bus = new class implements MessageBusInterface {
+            public function dispatch(object $message, array $stamps = []): Envelope
+            {
+                $envelope = Envelope::wrap($message, $stamps);
+                if ($envelope->last(FlushBatchHandlersStamp::class)) {
+                    throw new \RuntimeException('Flush failed');
+                }
+
+                return $envelope;
+            }
+        };
+
+        $dispatcher = new EventDispatcher();
+        $dispatcher->addListener(WorkerRunningEvent::class, function (WorkerRunningEvent $event) {
+            static $calls = 0;
+            if (++$calls >= 2) {
+                $event->getWorker()->stop();
+            }
+        });
+
+        $worker = new Worker(['transport' => $receiver], $bus, $dispatcher, clock: new MockClock());
+
+        $reflection = new \ReflectionClass($worker);
+        $unacksProperty = $reflection->getProperty('unacks');
+        $unacks = $unacksProperty->getValue($worker);
+        $dummyHandler = new DummyBatchHandler();
+        $envelopeWithNoAutoAck = $envelope->with(new NoAutoAckStamp(new HandlerDescriptor($dummyHandler)));
+        $unacks->attach($dummyHandler, [$envelopeWithNoAutoAck, 'transport']);
+
+        $worker->run();
+
+        $this->assertSame(1, $receiver->getRejectCount());
+        $rejectedEnvelopes = $receiver->getRejectedEnvelopes();
+        $this->assertCount(1, $rejectedEnvelopes);
+        $rejectedEnvelope = $rejectedEnvelopes[0];
+        $this->assertNull($rejectedEnvelope->last(NoAutoAckStamp::class));
+    }
 }
 
 class DummyQueueReceiver extends DummyReceiver implements QueueReceiverInterface

src/Symfony/Component/Messenger/Worker.php

@@ -257,9 +257,9 @@ private function flush(bool $force): bool
             [$envelope, $transportName] = $unacks[$batchHandler];
             try {
                 $this->bus->dispatch($envelope->with(new FlushBatchHandlersStamp($force)));
-                $envelope = $envelope->withoutAll(NoAutoAckStamp::class);
                 unset($unacks[$batchHandler], $batchHandler);
             } catch (\Throwable $e) {
+                $envelope = $envelope->withoutAll(NoAutoAckStamp::class);
                 $this->acks[] = [$transportName, $envelope, $e];
             }
         }