Close an IMAP conversation a worker has spent five minutes on

Both import consumers sat twenty minutes and more inside one fetch from
a server delivering a few kilobytes every twenty seconds (#95). The
thirty-second read timeout is per read and a trickle never trips it;
nothing timed the command. A consumer handles one message at a time, so
every other account's import waited behind it, and a stop signal did
nothing: a worker stops between messages.

ImapDeadline runs an alarm around every message a worker handles
(AbortStalledImap, for a received envelope only). When it goes off, the
IMAP connections opened since are shut down under their reader. The
read is at end-of-file at once, the library reports a lost connection,
and the message takes its retry ladder like any other network failure.

  not an exception   thrown from a signal handler it would land in the
  from the alarm     middle of a flush as readily as a read
  shutdown, not      webklex's logout() and destructor call
  fclose             stream_get_meta_data() on the stream and die with
                     a TypeError on a closed resource
  a total, not a     so it has to fit honest work: five minutes, which
  rate               is seconds for an ordinary page of fifty messages
                     and tight only for ~300 MB of attachments on a
                     slow line. Wrong, it shows as the class's warning
                     repeating for one folder

Checked against imap.gmail.com over TLS: after the shutdown a read
returns empty in under a millisecond with eof set.

The IDLE process holds its connection for twenty-nine minutes on
purpose and is not a worker handling a message; it is never armed.

Also the changelog for #94 and #96.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -61,6 +61,28 @@ domain keeps their name. No migration.
 cell falls back to when there is no room for the title stacked the two colours on top of each other
 in a few pixels of height. No migration.
 
+**An IMAP folder whose server closes idle connections no longer burns a CPU core.** Some servers
+(Lark, QQ) end an idling connection after a few minutes. plMail read that as a quiet mailbox and
+kept reading a connection that was gone, as fast as it could, until the 29-minute refresh — a core
+per folder. It now sees that the connection has ended and reconnects. No
+migration. [#94](https://github.com/karatektus/pl_mail/issues/94), reported by @montywei.
+
+**An import no longer sticks one message short of the end.** A server can list a message it will
+not hand over (iCloud does, for a draft deleted but not yet purged). Left alone on the last page,
+it made the import fail before its tries were counted, so it was retried every few minutes for good
+at 486 of 487 and the account's remaining folders waited behind it. It is given up on after five
+tries now. The same answer — nothing — from a folder that really is empty was also taken for the
+server not answering, so mail deleted from an emptied folder elsewhere stayed in plMail; an empty
+folder is read as empty. No
+migration. [#96](https://github.com/karatektus/pl_mail/issues/96), reported by @montywei.
+
+**One slow mail server can no longer hold up every import.** A server sending a trickle of data
+never tripped the 30-second read timeout, which is per read, so one fetch could occupy an import
+worker for as long as the trickle lasted while every other account waited. A worker now gives a
+message it is handling five minutes; after that its IMAP connections are closed, the message is
+retried later and the queue moves on. No
+migration. [#95](https://github.com/karatektus/pl_mail/issues/95), reported by @montywei.
+
 ## v0.3.4 — 2026-10-13
 
 **A Gmail import no longer runs into Google's quota and stays there.** Google lowered the

config/packages/messenger.yaml

@@ -267,6 +267,11 @@ framework:
                     # kind is dropped with a warning, the other is put back to
                     # wait, and neither is a CRITICAL any more.
                     - App\Infrastructure\Messaging\Middleware\GiveUpQuietlyOnThrottling
+                    # A clock on every message a worker handles, which closes
+                    # the IMAP connections it opened if it runs out — so one
+                    # server sending a trickle cannot hold a queue. See
+                    # App\Service\Imap\ImapDeadline.
+                    - App\Infrastructure\Messaging\Middleware\AbortStalledImap
 
         routing:
             # Admin "run now" buttons dispatch this; the scheduler dispatches

src/Domain/Helper/ImapConnectionFactory.php

@@ -2,6 +2,7 @@
 
 namespace App\Domain\Helper;
 
+use App\Service\Imap\ImapDeadline;
 use App\Entity\Mail\Account;
 use App\Domain\Enum\Account\AuthType;
 use App\Infrastructure\Imap\Utf8AwareMessageDecoder;
@@ -57,6 +58,7 @@ class ImapConnectionFactory
 
     public function __construct(
         private readonly OAuthTokenManager $tokenManager,
+        private readonly ?ImapDeadline $deadline = null,
     ) {
     }
 
@@ -107,6 +109,10 @@ class ImapConnectionFactory
 
         $client = new Client(self::config($accountConfig));
 
+        // Before connecting, not after: a server that accepts the socket and
+        // then says nothing is a stall too, and the greeting is read in here.
+        $this->deadline?->watch($client);
+
         $client->connect();
 
         return $client;

src/Infrastructure/Messaging/Middleware/AbortStalledImap.php

@@ -0,0 +1,47 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Infrastructure\Messaging\Middleware;
+
+use App\Service\Imap\ImapDeadline;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
+use Symfony\Component\Messenger\Middleware\StackInterface;
+use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+
+/**
+ * Runs the IMAP deadline around every message a worker handles.
+ *
+ * A middleware rather than a line in each handler, for the reason the
+ * deadline exists at all: the handler that stalled was not doing anything
+ * unusual, and the next one to stall will not be on anybody's list. See
+ * {@see ImapDeadline} for what expiring does and why.
+ *
+ * Only for a message that was RECEIVED — taken off a transport by a worker.
+ * A dispatch from a web request passes through here on its way to the queue
+ * and is not being handled; arming there would time a request, and would
+ * install a signal handler in a process that is not a worker.
+ */
+final readonly class AbortStalledImap implements MiddlewareInterface
+{
+    public function __construct(
+        private ImapDeadline $deadline,
+    ) {
+    }
+
+    public function handle(Envelope $envelope, StackInterface $stack): Envelope
+    {
+        if (null === $envelope->last(ReceivedStamp::class)) {
+            return $stack->next()->handle($envelope, $stack);
+        }
+
+        $this->deadline->arm();
+
+        try {
+            return $stack->next()->handle($envelope, $stack);
+        } finally {
+            $this->deadline->disarm();
+        }
+    }
+}

src/Service/Imap/ImapDeadline.php

@@ -0,0 +1,206 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Service\Imap;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Contracts\Service\ResetInterface;
+use Webklex\PHPIMAP\Client;
+use Webklex\PHPIMAP\Connection\Protocols\Protocol;
+use WeakReference;
+
+/**
+ * Ends an IMAP conversation that has gone on too long, so the worker holding
+ * it can do something else.
+ *
+ * ── What happened (#95) ─────────────────────────────────────────────────────
+ * Both import consumers sat for twenty minutes and more inside one fetch from
+ * Gmail's IMAP, which was delivering a few kilobytes every twenty seconds.
+ * Nothing was wrong enough to fail. There IS a read timeout, thirty seconds,
+ * but it is per read: a server that sends one byte inside every window never
+ * trips it, and no clock anywhere measured the whole command. A consumer
+ * handles one message at a time, so everything queued behind that fetch — the
+ * other accounts' imports, unrelated to Gmail — waited with it. Asking the
+ * consumer to stop did nothing either: it stops between messages, and it was
+ * not between messages.
+ *
+ * ── What this does ──────────────────────────────────────────────────────────
+ * While a worker handles a message there is an alarm running. If it goes off,
+ * every IMAP connection opened since is closed underneath whoever is reading
+ * it. The read fails at once, the handler fails the way it does when a
+ * network drops — which all of this code already expects — and Messenger puts
+ * the message back on its retry ladder, behind whatever was waiting.
+ *
+ * ── Why close the socket, rather than throw from the alarm ─────────────────
+ * A signal handler runs between any two statements of whatever was executing.
+ * An exception thrown from there would surface in the middle of a Doctrine
+ * flush as readily as in the middle of a read, and unwinding half a unit of
+ * work is how a worker ends up with a closed entity manager. Closing the
+ * socket changes nothing except for code that is talking to the server, and
+ * that code fails where it stands, at a line that has always been able to.
+ *
+ * It also matters for what comes next. Every caller disconnects in a
+ * `finally`, and a polite logout reads to the server's goodbye — through the
+ * same megabytes of trickle that were the problem. A socket that has been shut down has
+ * no goodbye to wait for.
+ *
+ * ── The length ──────────────────────────────────────────────────────────────
+ * Five minutes. It is a total and not a rate — nothing here can see how fast
+ * a connection is moving, only how long the message has been in hand — so it
+ * has to leave room for honest work, and the largest honest unit is one import
+ * page: fifty messages. That is seconds for ordinary mail and a few minutes
+ * for a page of large attachments on a slow line.
+ *
+ * It was fifteen at first, on the reasoning that a wrong cut costs a page
+ * fetched again. But every minute of it is a minute all the other accounts
+ * wait, and fifteen of them per stalled message is most of the stall the
+ * report was about. What five risks instead is a page that genuinely needs
+ * longer — three hundred megabytes of attachments at under eight megabits a
+ * second — being cut every time it is tried. That shows as this class's
+ * warning in the log, repeatedly, for the same folder; it is the sign to
+ * raise this.
+ *
+ * Only where there is a worker and the means: `pcntl`, and a message that was
+ * received from a transport. A web request, a console command and the IDLE
+ * process — which holds its connection for twenty-nine minutes on purpose —
+ * never arm it.
+ */
+final class ImapDeadline implements ResetInterface
+{
+    public const int SECONDS = 300;
+
+    /** @var list<WeakReference<Client>> */
+    private array $clients = [];
+
+    /** How many handlers deep the worker is; only the outermost owns the alarm. */
+    private int $depth = 0;
+
+    private bool $installed = false;
+
+    public function __construct(
+        private readonly LoggerInterface $logger,
+        private readonly int $seconds = self::SECONDS,
+    ) {
+    }
+
+    /** A connection that an expiring deadline should close. Held weakly: this must not keep one alive. */
+    public function watch(Client $client): void
+    {
+        $this->clients = array_values(array_filter(
+            $this->clients,
+            static fn (WeakReference $reference): bool => null !== $reference->get(),
+        ));
+
+        $this->clients[] = WeakReference::create($client);
+    }
+
+    public static function available(): bool
+    {
+        return 'cli' === PHP_SAPI && true === function_exists('pcntl_alarm') && true === function_exists('pcntl_async_signals');
+    }
+
+    /** Start the clock for one handled message. Nested handling shares the outermost clock. */
+    public function arm(): void
+    {
+        if (false === self::available() || $this->seconds <= 0) {
+            return;
+        }
+
+        if (1 !== ++$this->depth) {
+            return;
+        }
+
+        if (false === $this->installed) {
+            pcntl_async_signals(true);
+            pcntl_signal(SIGALRM, function (): void {
+                $this->expire();
+            });
+
+            $this->installed = true;
+        }
+
+        // Only what is opened from here on belongs to this message.
+        $this->clients = [];
+
+        pcntl_alarm($this->seconds);
+    }
+
+    public function disarm(): void
+    {
+        if (false === self::available() || $this->seconds <= 0) {
+            return;
+        }
+
+        if (0 !== --$this->depth) {
+            $this->depth = max(0, $this->depth);
+
+            return;
+        }
+
+        pcntl_alarm(0);
+        $this->clients = [];
+    }
+
+    /**
+     * Close every watched connection under whoever is using it.
+     *
+     * Public because it is the whole effect of the class and has to be
+     * testable without waiting five minutes for a signal.
+     *
+     * @return int how many connections were closed
+     */
+    public function expire(): int
+    {
+        $closed = 0;
+
+        foreach ($this->clients as $reference) {
+            $client = $reference->get();
+
+            if (null === $client) {
+                continue;
+            }
+
+            try {
+                $connection = $client->getConnection();
+                $stream     = $connection instanceof Protocol ? $connection->getStream() : null;
+
+                if (true === is_resource($stream)) {
+                    // Shut down, not fclose()d. Both end the conversation;
+                    // only this one leaves the library holding something it
+                    // can still ask questions of. With the resource closed
+                    // outright, its logout() — which every caller's `finally`
+                    // reaches, and its destructor after them — dies on
+                    // stream_get_meta_data() with a TypeError, from a
+                    // destructor, wherever the garbage collector happens to
+                    // be. Shut down, every read is simply at end-of-file: the
+                    // library reports a lost connection, as it does when a
+                    // network drops, and tidies up after itself.
+                    if (false === @stream_socket_shutdown($stream, STREAM_SHUT_RDWR)) {
+                        fclose($stream);
+                    }
+
+                    ++$closed;
+                }
+            } catch (\Throwable) {
+                // Already gone, or never connected: nothing is being read from it.
+            }
+        }
+
+        $this->clients = [];
+
+        if ($closed > 0) {
+            $this->logger->warning('An IMAP conversation outlasted its deadline and was closed', [
+                'seconds'     => $this->seconds,
+                'connections' => $closed,
+            ]);
+        }
+
+        return $closed;
+    }
+
+    public function reset(): void
+    {
+        $this->clients = [];
+    }
+}

tests/Service/Imap/ImapDeadlineTest.php

@@ -0,0 +1,254 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Tests\Service\Imap;
+
+use App\Infrastructure\Messaging\Middleware\AbortStalledImap;
+use App\Service\Imap\ImapDeadline;
+use PHPUnit\Framework\TestCase;
+use Psr\Log\NullLogger;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
+use Symfony\Component\Messenger\Middleware\StackInterface;
+use Symfony\Component\Messenger\Stamp\ReceivedStamp;
+use Webklex\PHPIMAP\Client;
+use Webklex\PHPIMAP\Config;
+use Webklex\PHPIMAP\Connection\Protocols\ImapProtocol;
+use Webklex\PHPIMAP\Connection\Protocols\ProtocolInterface;
+use Webklex\PHPIMAP\Connection\Protocols\Response;
+use Webklex\PHPIMAP\Exceptions\RuntimeException as ImapRuntimeException;
+
+/**
+ * A stalled IMAP conversation is cut off, and only that.
+ *
+ * #95: two consumers held for twenty minutes by one server sending a trickle,
+ * and every other account's import waiting behind them. The per-read timeout
+ * never fires on a trickle, so the deadline is on the whole handled message.
+ *
+ * Real sockets and a real alarm. The claim is that a read which would have
+ * gone on waiting stops, and the only way to know is to be waiting on one.
+ */
+final class ImapDeadlineTest extends TestCase
+{
+    /** @var list<resource> */
+    private array $peers = [];
+
+    protected function tearDown(): void
+    {
+        if (true === ImapDeadline::available()) {
+            pcntl_alarm(0);
+        }
+
+        foreach ($this->peers as $peer) {
+            if (true === is_resource($peer)) {
+                fclose($peer);
+            }
+        }
+    }
+
+    public function testExpiringClosesTheConnectionUnderItsReader(): void
+    {
+        $deadline = new ImapDeadline(new NullLogger());
+        $client   = $this->client();
+
+        $deadline->watch($client);
+
+        self::assertSame(1, $deadline->expire());
+
+        // Still a stream the library can ask questions of — its logout and its
+        // destructor both do — but one with nothing more to say.
+        self::assertTrue(is_resource($this->streamOf($client)));
+
+        $started = microtime(true);
+
+        try {
+            $this->protocolOf($client)->nextLine(new Response(0, false));
+            self::fail('a read on a conversation that was cut off cannot succeed');
+        } catch (ImapRuntimeException $e) {
+            // What the library says for a connection that dropped, which is
+            // the failure every caller already knows how to have.
+            self::assertSame('empty response', $e->getMessage());
+        }
+
+        self::assertLessThan(1.0, microtime(true) - $started, 'at once, not after another read timeout');
+        self::assertTrue(feof($this->streamOf($client)));
+    }
+
+    public function testAConnectionNobodyHoldsAnyMoreIsNotKeptAliveToBeClosed(): void
+    {
+        $deadline = new ImapDeadline(new NullLogger());
+        $client   = $this->client();
+
+        $deadline->watch($client);
+        unset($client);
+        gc_collect_cycles();
+
+        self::assertSame(0, $deadline->expire());
+    }
+
+    /** The whole thing, on the clock: a read that had ten seconds to wait is over in about one. */
+    public function testAHandlerStillReadingWhenTheClockRunsOutIsStopped(): void
+    {
+        if (false === ImapDeadline::available()) {
+            self::markTestSkipped('needs pcntl');
+        }
+
+        $deadline   = new ImapDeadline(new NullLogger(), seconds: 1);
+        $middleware = new AbortStalledImap($deadline);
+        $client     = $this->client();
+        $started    = microtime(true);
+        $failure    = null;
+
+        $handler = new class ($deadline, $client, $this) implements MiddlewareInterface {
+            public function __construct(
+                private readonly ImapDeadline $deadline,
+                private readonly Client $client,
+                private readonly ImapDeadlineTest $test,
+            ) {
+            }
+
+            public function handle(Envelope $envelope, StackInterface $stack): Envelope
+            {
+                // As ImapConnectionFactory does for a connection a handler opens.
+                $this->deadline->watch($this->client);
+
+                $stream = $this->test->streamOf($this->client);
+                stream_set_timeout($stream, 0, 200_000);
+
+                // A server that never sends the line. Each read gives up after
+                // its own short timeout, which is what the per-read timeout
+                // looks like from inside: on its own it ends nothing, and this
+                // loop would run its full ten seconds.
+                $protocol = $this->test->protocolOf($this->client);
+
+                for ($i = 0; $i < 50; ++$i) {
+                    if (true === feof($stream)) {
+                        // The library's own words for a connection that is gone.
+                        $protocol->nextLine(new Response(0, false));
+                    }
+
+                    fread($stream, 1);
+                }
+
+                return $envelope;
+            }
+        };
+
+        try {
+            $middleware->handle(
+                new Envelope(new \stdClass(), [new ReceivedStamp('ingest')]),
+                $this->stackOf($handler),
+            );
+        } catch (\Throwable $e) {
+            $failure = $e;
+        }
+
+        self::assertNotNull($failure, 'the read was cut off rather than left to finish');
+        self::assertLessThan(4.0, microtime(true) - $started, 'at the deadline, not at the end of the ten seconds');
+        self::assertSame(0, pcntl_alarm(0), 'and the clock was stopped on the way out');
+    }
+
+    /** Passing through on its way to a queue is not being handled, and is not timed. */
+    public function testAMessageThatWasNotReceivedByAWorkerIsNotTimed(): void
+    {
+        if (false === ImapDeadline::available()) {
+            self::markTestSkipped('needs pcntl');
+        }
+
+        $handler = new class () implements MiddlewareInterface {
+            public ?int $secondsLeft = null;
+
+            public function handle(Envelope $envelope, StackInterface $stack): Envelope
+            {
+                // pcntl_alarm() answers with the seconds left on the old alarm.
+                $this->secondsLeft = pcntl_alarm(0);
+
+                return $envelope;
+            }
+        };
+
+        new AbortStalledImap(new ImapDeadline(new NullLogger()))->handle(new Envelope(new \stdClass()), $this->stackOf($handler));
+
+        self::assertSame(0, $handler->secondsLeft);
+
+        new AbortStalledImap(new ImapDeadline(new NullLogger()))->handle(
+            new Envelope(new \stdClass(), [new ReceivedStamp('ingest')]),
+            $this->stackOf($handler),
+        );
+
+        self::assertGreaterThan(0, $handler->secondsLeft, 'while a received one has the clock running');
+    }
+
+    // ── helpers ──────────────────────────────────────────────────────────────
+
+    /** @return resource */
+    public function streamOf(Client $client): mixed
+    {
+        return $this->protocolOf($client)->getStream();
+    }
+
+    public function protocolOf(Client $client): ImapProtocol
+    {
+        $protocol = $client->getConnection();
+
+        self::assertInstanceOf(ImapProtocol::class, $protocol);
+
+        return $protocol;
+    }
+
+    private function stackOf(MiddlewareInterface $next): StackInterface
+    {
+        return new class ($next) implements StackInterface {
+            public function __construct(private readonly MiddlewareInterface $next)
+            {
+            }
+
+            public function next(): MiddlewareInterface
+            {
+                return $this->next;
+            }
+        };
+    }
+
+    /** A client whose connection is one end of a socket pair nothing is ever written to. */
+    private function client(): Client
+    {
+        $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+
+        self::assertIsArray($pair);
+
+        $this->peers[] = $pair[1];
+
+        $protocol = new class (Config::make(), $pair[0]) extends ImapProtocol {
+            /** @param resource $socket */
+            public function __construct(Config $config, mixed $socket)
+            {
+                parent::__construct($config);
+
+                $this->stream = $socket;
+            }
+
+            /**
+             * The real one logs out, and a logout waits for the server's
+             * goodbye — thirty seconds of it, from a socket pair with nobody
+             * on the other end, once per test.
+             */
+            public function __destruct()
+            {
+            }
+        };
+
+        return new class (Config::make(), $protocol) extends Client {
+            public function __construct(Config $config, private readonly ProtocolInterface $fake)
+            {
+                parent::__construct($config);
+            }
+
+            public function getConnection(): ProtocolInterface
+            {
+                return $this->fake;
+            }
+        };
+    }
+}