fix(sync): two mails about one parcel in a batch no longer stop the sync

A fresh install with one GMX mailbox logged, during its first sync:

  Failed to build message; it is asked for again next sync
  "error": "The EntityManager is closed."

The first sync reads the INBOX fifty messages at a time, and
ExtractInsightsStep harvests the whole batch before one flush. Two mails
carrying the same fact (the shipping and the delivery notice for one
parcel) each looked for the card in the table, found nothing, and each
inserted. The flush broke uniq_mail_insight_account_dedupe, Doctrine
closed the entity manager, PostIngestPipeline logged the step as failed
and went on, and every persist of the next batch threw
EntityManagerClosed, blamed on its own message. The folder's sync then
died on the flush after it.

  MailInsightRepository  findOneByDedupe() also finds a row persisted
                         earlier in the same unit of work, so the
                         second mail refreshes the first one's card.
  MessageSyncer          reopens a closed manager between batches;
                         clear() alone left it closed.
  ImapAccountSyncer      reopens it after a failed folder, and finds
                         each folder again by id before syncing it.

That last lookup fixes a bug found on the way. MessageSyncer clears the
entity manager between batches, so in an account sync every folder
after the first one with new mail was handed over detached, and what
the sweep wrote onto it (sweptAt, the folder's UIDVALIDITY) was flushed
into nothing. Without a recorded UIDVALIDITY, a folder rebuilt on the
server is not recognised as rebuilt.

No mail was lost: a failed batch was never committed, the mark did not
move past it, and the attempt counter was never saved either. The cards
of the failed batches were lost; app:backfill insights brings them back.

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

src/Repository/Insight/MailInsightRepository.php

@@ -18,9 +18,31 @@ class MailInsightRepository extends ServiceEntityRepository
         parent::__construct($registry, MailInsight::class);
     }
 
-    /** The upsert's read half — see InsightHarvester. */
+    /**
+     * The upsert's read half — see InsightHarvester.
+     *
+     * A row persisted in this unit of work and not flushed yet counts as much
+     * as one in the table. The sync harvests a whole batch and flushes once
+     * (ExtractInsightsStep), so when the shipping and the delivery mail for one
+     * parcel arrived in the same batch, each found nothing in the table and
+     * each inserted, and the one flush broke uniq_mail_insight_account_dedupe.
+     * That closed the entity manager under the rest of the sync. A first sync
+     * is thousands of mails in batches of fifty, so a fresh install hit it
+     * within minutes.
+     *
+     * Scanned rather than queried: a pending row has no id and is not in the
+     * table, so the unit of work is the only place it can be found.
+     */
     public function findOneByDedupe(Account $account, string $dedupeKey): ?MailInsight
     {
+        foreach ($this->getEntityManager()->getUnitOfWork()->getScheduledEntityInsertions() as $pending) {
+            if ($pending instanceof MailInsight
+                && $dedupeKey === $pending->dedupeKey
+                && $account->id === $pending->account->id) {
+                return $pending;
+            }
+        }
+
         return $this->findOneBy(['account' => $account, 'dedupeKey' => $dedupeKey]);
     }
 

src/Service/Imap/ImapAccountSyncer.php

@@ -7,7 +7,10 @@ namespace App\Service\Imap;
 use App\Domain\Helper\ImapConnectionFactory;
 use App\Domain\Interface\AccountSyncerInterface;
 use App\Entity\Mail\Account;
+use App\Entity\Mail\Mailbox;
 use App\Repository\Mail\MailboxRepository;
+use Doctrine\ORM\EntityManagerInterface;
+use Doctrine\Persistence\ManagerRegistry;
 use Psr\Log\LoggerInterface;
 
 /**
@@ -24,6 +27,8 @@ final readonly class ImapAccountSyncer implements AccountSyncerInterface
         private MessageSyncer         $messageSyncer,
         private ImapConnectionFactory $imapConnectionFactory,
         private LoggerInterface       $logger,
+        private EntityManagerInterface $em,
+        private ManagerRegistry       $registry,
     ) {}
 
     public function supports(Account $account): bool
@@ -76,20 +81,42 @@ final readonly class ImapAccountSyncer implements AccountSyncerInterface
             return [];
         }
 
+        $mailboxIds       = array_map(static fn (Mailbox $mailbox): int => (int) $mailbox->id, $mailboxes);
         $client           = $this->imapConnectionFactory->connect($account);
         $syncedMailboxIds = [];
 
         try {
-            foreach ($mailboxes as $mailbox) {
+            foreach ($mailboxIds as $mailboxId) {
+                // Found again for each folder rather than taken from the list
+                // above. MessageSyncer clears the entity manager between
+                // batches, so once one folder had new mail every later entry
+                // in that list was detached, and what the sweep wrote onto it
+                // (sweptAt, the folder's UIDVALIDITY) was flushed into
+                // nothing. A folder that never records its UIDVALIDITY cannot
+                // be recognised as rebuilt on the server.
+                $mailbox = $this->mailboxRepository->find($mailboxId);
+
+                if (null === $mailbox) {
+                    continue;
+                }
+
                 try {
                     $this->messageSyncer->syncMailbox($mailbox, $client);
-                    $syncedMailboxIds[] = $mailbox->id;
+                    $syncedMailboxIds[] = $mailboxId;
                 } catch (\Throwable $e) {
                     $this->logger->error('ImapAccountSyncer: mailbox sync failed', [
-                        'mailboxId' => $mailbox->id,
+                        'mailboxId' => $mailboxId,
                         'error'     => $e->getMessage(),
                         'exception' => $e,
                     ]);
+
+                    // A failure out of a flush closes the manager, and the next
+                    // folder would die on its first persist, with every one of
+                    // its messages blamed for it. Reopened, one folder's failure
+                    // stays that folder's.
+                    if (false === $this->em->isOpen()) {
+                        $this->registry->resetManager();
+                    }
                 }
             }
         } finally {

src/Service/Imap/MessageSyncer.php

@@ -22,6 +22,7 @@ use App\Service\Mail\MisfiledBodyUnpacker;
 use App\Service\Mail\PostIngestPipeline;
 use DateTimeImmutable;
 use Doctrine\ORM\EntityManagerInterface;
+use Doctrine\Persistence\ManagerRegistry;
 use Psr\Log\LoggerInterface;
 use Webklex\PHPIMAP\Attribute;
 use Webklex\PHPIMAP\Client;
@@ -53,6 +54,7 @@ class MessageSyncer
         private readonly ImapConnectionFactory $connections,
         private readonly VanishedMessageReconciler $vanished,
         private readonly GhostMessageReaper $ghosts,
+        private readonly ManagerRegistry $registry,
     ) {}
 
     /**
@@ -151,7 +153,7 @@ class MessageSyncer
                 ->chunked(function ($batch) use ($mailboxId, $accountId, $lastSeenUid, &$synced, &$syncedUids, &$lowestSkippedUid, $presence) {
                     $this->processBatch($batch, $mailboxId, $accountId, $lastSeenUid, $syncedUids, $lowestSkippedUid, $presence);
                     $synced += count($batch);
-                    $this->em->clear();
+                    $this->startNextBatch();
                     $this->logger->info(sprintf('Synced %d messages so far', $synced));
                 }, self::BATCH_SIZE);
 
@@ -349,6 +351,34 @@ class MessageSyncer
         $this->postIngest->run($mailbox->account, $ingested);
     }
 
+    /**
+     * Drop the finished batch's unit of work, and reopen the manager if
+     * something closed it.
+     *
+     * Post-ingest steps run after the batch is committed and are allowed to
+     * fail: PostIngestPipeline logs the failure and carries on, because the
+     * mail is already stored. But Doctrine closes the manager when a flush
+     * fails, and clear() does not reopen it. So one step that failed a flush
+     * made the next batch throw EntityManagerClosed on its first persist.
+     * Every message in that batch was logged as "Failed to build message"
+     * though nothing was wrong with any of them, and the folder's sync then
+     * died on the flush after.
+     *
+     * Resetting is safe here for the same reason clearing is: nothing that
+     * writes holds an entity across a batch, and processBatch() finds the
+     * mailbox again by id.
+     */
+    private function startNextBatch(): void
+    {
+        if (true === $this->em->isOpen()) {
+            $this->em->clear();
+
+            return;
+        }
+
+        $this->registry->resetManager();
+    }
+
     /**
      * Keep a UID that could not be stored inside the next sync's range — up to
      * a point.

tests/Service/Imap/ImapAccountSyncerTest.php

@@ -0,0 +1,149 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Tests\Service\Imap;
+
+use App\Domain\Helper\ImapConnectionFactory;
+use App\Entity\Mail\Mailbox;
+use App\Repository\Mail\MailboxRepository;
+use App\Service\Imap\ImapAccountSyncer;
+use App\Service\Imap\MailboxSyncer;
+use App\Service\Imap\MessageSyncer;
+use App\Tests\Support\Mail\SeedsMarkerFixtures;
+use Doctrine\DBAL\Connection;
+use Doctrine\ORM\EntityManagerInterface;
+use Doctrine\Persistence\ManagerRegistry;
+use Psr\Log\NullLogger;
+use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
+
+/**
+ * One folder's trouble stays that folder's.
+ *
+ * An account sync walks its folders with one entity manager, and two things a
+ * folder's sync does to that manager used to reach the folders after it. A
+ * failed flush closes it: every later folder then failed on its first persist,
+ * each of its messages logged as unbuildable. And MessageSyncer clears it
+ * between batches: every later folder was then handed over detached, and what
+ * the sweep wrote onto it was flushed into nothing.
+ *
+ * The folder sync is a stand-in that does to the manager exactly what the real
+ * one does. Both faults live between folders, which is this class, and the real
+ * one could not get there without an IMAP server.
+ */
+final class ImapAccountSyncerTest extends KernelTestCase
+{
+    use SeedsMarkerFixtures;
+
+    private Connection $connection;
+
+    protected function setUp(): void
+    {
+        self::bootKernel();
+
+        $this->em         = static::getContainer()->get(EntityManagerInterface::class);
+        $this->connection = static::getContainer()->get(Connection::class);
+
+        $this->connection->beginTransaction();
+
+        $this->user    = $this->seedUser();
+        $this->account = $this->seedAccount();
+
+        $this->seedFolder('Archive');
+        $this->seedFolder('INBOX');
+        $this->em->flush();
+    }
+
+    protected function tearDown(): void
+    {
+        if (true === $this->connection->isTransactionActive()) {
+            $this->connection->rollBack();
+        }
+
+        parent::tearDown();
+    }
+
+    public function testAFolderWhoseFlushFailedDoesNotFailTheNextOne(): void
+    {
+        $handedOver = $this->syncAccount(function (): void {
+            // What a failed flush leaves behind: UnitOfWork closes the manager
+            // on its way out, and then the error escapes.
+            $this->em->close();
+
+            throw new \RuntimeException('SQLSTATE[23505]: Unique violation');
+        });
+
+        self::assertTrue($handedOver[1]['open'], 'the next folder was handed a closed entity manager');
+    }
+
+    public function testEveryFolderIsHandedOverManagedAfterAnEarlierOneClearedTheManager(): void
+    {
+        // What any folder with new mail does between its batches.
+        $handedOver = $this->syncAccount(fn () => $this->em->clear());
+
+        self::assertTrue($handedOver[1]['managed'], 'the next folder was detached, so its sweep wrote nothing');
+    }
+
+    /**
+     * Syncs the account through a real ImapAccountSyncer. The first folder's
+     * sync does $first; every folder reports the state it arrived in.
+     *
+     * In call order rather than by id: the folders come from an unordered
+     * findBy(), and which of the two goes first is not this test's business.
+     *
+     * @return list<array{open: bool, managed: bool}>
+     */
+    private function syncAccount(\Closure $first): array
+    {
+        $handedOver = [];
+
+        $folders = $this->createStub(MessageSyncer::class);
+        $folders->method('syncMailbox')->willReturnCallback(
+            function (Mailbox $mailbox) use ($first, &$handedOver): void {
+                $handedOver[] = [
+                    'open'    => $this->em->isOpen(),
+                    'managed' => $this->em->contains($mailbox),
+                ];
+
+                if (1 === count($handedOver)) {
+                    $first();
+                }
+            },
+        );
+
+        $structure = $this->createStub(MailboxSyncer::class);
+        $structure->method('syncForAccount')->willReturn(
+            ['created' => 0, 'updated' => 0, 'deleted' => 0, 'renamed' => 0, 'missing' => 0],
+        );
+
+        $connections = $this->createStub(ImapConnectionFactory::class);
+        $connections->method('connect')->willReturn(new FakeListingClient([]));
+
+        $syncer = new ImapAccountSyncer(
+            static::getContainer()->get(MailboxRepository::class),
+            $structure,
+            $folders,
+            $connections,
+            new NullLogger(),
+            $this->em,
+            static::getContainer()->get(ManagerRegistry::class),
+        );
+
+        $syncer->sync($this->account);
+
+        self::assertCount(2, $handedOver, 'both folders were synced');
+
+        return $handedOver;
+    }
+
+    private function seedFolder(string $name): void
+    {
+        $mailbox                = new Mailbox();
+        $mailbox->account       = $this->account;
+        $mailbox->name          = $name;
+        $mailbox->fullPath      = $name;
+        $mailbox->isSyncEnabled = true;
+
+        $this->em->persist($mailbox);
+    }
+}

tests/Service/Insight/InsightHarvesterTest.php

@@ -86,6 +86,32 @@ final class InsightHarvesterTest extends KernelTestCase
         self::assertSame('stub:123', $rows[0]->dedupeKey, 'the harvester scopes the key by extractor');
     }
 
+    public function testTwoMailsAboutOneParcelInOneBatchAreOneRowRatherThanAFailedFlush(): void
+    {
+        // Both mails exist before either is harvested, and nothing is flushed
+        // between them: that is a sync batch. ExtractInsightsStep harvests all
+        // of it and flushes once, so the second mail cannot find the first
+        // one's row in the table.
+        $first  = $this->thread('shipped')->messages->first();
+        $second = $this->thread('delivered')->messages->first();
+
+        $this->harvester($this->stub([
+            new InsightDraft(InsightKind::Parcel, 'DHL · 123', '123', ['status' => 'in_transit']),
+        ]))->harvest($first);
+
+        $this->harvester($this->stub([
+            new InsightDraft(InsightKind::Parcel, 'DHL · 123', '123', ['status' => 'delivered']),
+        ]))->harvest($second);
+
+        $this->em->flush();
+
+        $rows = $this->repository()->findBy(['account' => $this->account]);
+
+        self::assertCount(1, $rows, 'two mails about one parcel in one batch are one card');
+        self::assertSame('delivered', $rows[0]->payload['status'], 'the newest statement wins');
+        self::assertSame($second->id, $rows[0]->message?->id, 'the link follows the newest mail');
+    }
+
     public function testDismissalOutlivesTheNextUpsert(): void
     {
         $draft = new InsightDraft(InsightKind::Parcel, 'DHL · 9', '9', []);