Abandon all pending operations when connection was closed by the server

composer.json

@@ -84,7 +84,7 @@
         "fixcs": "tools/php-cs-fixer/vendor/bin/php-cs-fixer fix --diff --verbose",
         "infection": "tools/infection/vendor/bin/infection --show-mutations",
         "normalize": "@composer bin composer-normalize normalize --diff ../../composer.json",
-        "phpstan": "tools/phpstan/vendor/bin/phpstan analyze",
+        "phpstan": "tools/phpstan/vendor/bin/phpstan analyze --memory-limit=512M",
         "pre-command-run": "mkdir -p var",
         "psalm": "tools/psalm/vendor/bin/psalm --show-info --no-diff --no-cache",
         "rector": "tools/rector/vendor/bin/rector process",

src/Channel.php

@@ -284,6 +284,14 @@ public function consumeIterator(
         /** @var Internal\QueueIterator<DeliveryMessage> $iterator */
         $iterator = Internal\QueueIterator::buffered($consumerTag, $this, $size);
 
+        $canceller = new Canceller(
+            $iterator->complete(...),
+            $iterator->cancel(...),
+            $iterator->abandon(...),
+        );
+
+        $this->cancellations->add($consumerTag, $canceller);
+
         $this->consume(
             callback: $iterator->push(...),
             queue: $queue,
@@ -339,6 +347,7 @@ public function consumeBatch(
         $canceller = new Canceller(
             $iterator->complete(...),
             $iterator->cancel(...),
+            $iterator->abandon(...),
         );
 
         $this->cancellations->add($consumerTag, $canceller);
@@ -782,7 +791,7 @@ public function abandon(\Throwable $e): void
     {
         $this->hooks->reject($this->channelId, $e);
         $this->hooks->unsubscribe($this->channelId);
-        $this->cancellations->cancelAll(error: $e);
+        $this->cancellations->abandon($e);
         $this->supervisor->stop();
         $this->closed = new Sync\Once(static fn(): bool => true);
     }

src/Internal/Cancellation/CancellationStorage.php

@@ -33,6 +33,17 @@ public function cancel(string $consumerTag, bool $noWait = false, ?\Throwable $e
         }
     }
 
+    public function abandon(\Throwable $e): void
+    {
+        try {
+            foreach ($this->cancellers as $canceller) {
+                $canceller->abandon($e);
+            }
+        } finally {
+            $this->cancellers = [];
+        }
+    }
+
     public function cancelAll(bool $noWait = false, ?\Throwable $error = null): void
     {
         try {

src/Internal/Cancellation/Canceller.php

@@ -19,10 +19,12 @@ final class Canceller
     /**
      * @param callable(bool): void $complete
      * @param callable(\Throwable, bool): void $cancel
+     * @param callable(\Throwable): void $abandon
      */
     public function __construct(
         private readonly mixed $complete,
         private readonly mixed $cancel,
+        private readonly mixed $abandon,
     ) {
         $this->deferred = new DeferredCancellation();
     }
@@ -46,6 +48,20 @@ public function cancel(bool $noWait = false, ?\Throwable $e = null): void
         }
     }
 
+    public function abandon(\Throwable $e): void
+    {
+        if ($this->cancelled) {
+            return;
+        }
+
+        try {
+            ($this->abandon)($e);
+            $this->deferred->cancel();
+        } finally {
+            $this->cancelled = true;
+        }
+    }
+
     public function cancellation(): Cancellation
     {
         return $this->deferred->getCancellation();

src/Internal/QueueIterator.php

@@ -56,14 +56,25 @@ public function push(mixed $delivery): void
 
     public function complete(bool $noWait = false): void
     {
-        $this->channel->cancel($this->consumerTag, $noWait);
-        $this->queue->complete();
+        if (!$this->queue->isComplete()) {
+            $this->queue->complete();
+            $this->channel->cancel($this->consumerTag, $noWait);
+        }
     }
 
     public function cancel(\Throwable $e, bool $noWait = false): void
     {
-        $this->channel->cancel($this->consumerTag, $noWait);
-        $this->queue->error($e);
+        if (!$this->queue->isComplete()) {
+            $this->queue->error($e);
+            $this->channel->cancel($this->consumerTag, $noWait);
+        }
+    }
+
+    public function abandon(\Throwable $e): void
+    {
+        if (!$this->queue->isComplete()) {
+            $this->queue->error($e);
+        }
     }
 
     public function continue(?Cancellation $cancellation = null): bool

src/Iterator.php

@@ -24,6 +24,8 @@ public function complete(bool $noWait = false): void;
      */
     public function cancel(\Throwable $e, bool $noWait = false): void;
 
+    public function abandon(\Throwable $e): void;
+
     /**
      * @throws CancelledException
      */