fix(jmap): state tokens follow commit order per account

The change log's identity value is the state token and is drawn at
INSERT, so two writers of one account could commit out of order and a
client holding the later token never saw the earlier change. A BEFORE
INSERT trigger now takes a per-account advisory xact lock and draws the
number after it. A trigger because Doctrine fires no event inside its
flush transaction before the inserts.

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

migrations/Version20260923140100.php

@@ -0,0 +1,76 @@
+<?php
+
+declare(strict_types=1);
+
+namespace DoctrineMigrations;
+
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\Migrations\AbstractMigration;
+
+/**
+ * JMAP state tokens in commit order, per account.
+ *
+ * The change log's identity column is the state token, and an identity value
+ * is handed out at INSERT, not at COMMIT. Two flushes writing change rows for
+ * one account could therefore commit out of order: A takes 10, B takes 11 and
+ * commits, a client reads state 11 — and when A commits, row 10 sits below a
+ * token the client already holds and /changes never reports it. The change is
+ * lost to that client for good, which is exactly what state tokens exist to
+ * prevent.
+ *
+ * The fix is to serialise the writers of one account from the moment they
+ * take a number until they commit: a transaction-scoped advisory lock keyed by
+ * the account, taken BEFORE the number. That has to be a trigger rather than a
+ * repository call from PHP. Doctrine opens the flush transaction itself and
+ * fires no event inside it before the inserts; onFlush runs before BEGIN, so a
+ * lock taken there would be released by autocommit on the spot. And the
+ * column default is evaluated before a BEFORE trigger sees the row, so the
+ * trigger cannot merely lock — it locks and then draws the number again.
+ * Doctrine reads the id back with LASTVAL(), which is the trigger's nextval,
+ * so the entity carries the number that was stored. The default's own draw is
+ * a gap, and gaps in this sequence are already normal (it is shared by every
+ * account and object type).
+ *
+ * Cost: one uncontended lock per row, re-entrant within a transaction. Writers
+ * of different accounts never wait for each other. A flush touching several
+ * accounts (Gmailify attribution) takes several locks in row order; two such
+ * flushes in opposite order can deadlock, which Postgres detects and resolves
+ * by failing one, and the transport retries it.
+ *
+ * Key: the two-int form, (JMAP, account_id), so it cannot collide with the
+ * single-bigint keys used by MigrateCommand and UserRepository. 1246576976 is
+ * the bytes of "JMAP" as an int4.
+ */
+final class Version20260923140100 extends AbstractMigration
+{
+    public function getDescription(): string
+    {
+        return 'Serialise jmap_change_log writers per account so state tokens follow commit order.';
+    }
+
+    public function up(Schema $schema): void
+    {
+        $this->addSql(<<<'SQL'
+            CREATE OR REPLACE FUNCTION plmail_jmap_change_log_sequence() RETURNS trigger
+            LANGUAGE plpgsql AS $$
+            BEGIN
+                PERFORM pg_advisory_xact_lock(1246576976, NEW.account_id);
+                NEW.sequence := nextval(pg_get_serial_sequence('jmap_change_log', 'sequence'));
+                RETURN NEW;
+            END
+            $$
+        SQL);
+
+        $this->addSql(<<<'SQL'
+            CREATE OR REPLACE TRIGGER jmap_change_log_sequence
+            BEFORE INSERT ON jmap_change_log
+            FOR EACH ROW EXECUTE FUNCTION plmail_jmap_change_log_sequence()
+        SQL);
+    }
+
+    public function down(Schema $schema): void
+    {
+        $this->addSql('DROP TRIGGER IF EXISTS jmap_change_log_sequence ON jmap_change_log');
+        $this->addSql('DROP FUNCTION IF EXISTS plmail_jmap_change_log_sequence()');
+    }
+}

src/Jmap/State/ChangeLog.php

@@ -13,6 +13,11 @@ use Doctrine\ORM\Mapping as ORM;
  * the highest sequence recorded for it, and "/changes" simply returns rows
  * with sequence > sinceState.
  *
+ * That only holds if a later number never commits before an earlier one of the
+ * same account, and an identity value is drawn at INSERT, not COMMIT. A BEFORE
+ * INSERT trigger (Version20260923140100) takes a per-account advisory lock and
+ * then draws the number, so one account's writers queue until commit.
+ *
  * accountId is stored as a scalar (not a ManyToOne) on purpose: these rows are
  * written from long-running sync handlers where holding entity references
  * across flush() is the documented footgun. A plain id sidesteps it entirely.

tests/Jmap/State/ChangeLogCommitOrderTest.php

@@ -0,0 +1,106 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Tests\Jmap\State;
+
+use App\Jmap\State\ChangeLog;
+use App\Jmap\State\JmapObjectType;
+use App\Jmap\State\StateManager;
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\DriverManager;
+use Doctrine\DBAL\Schema\Schema;
+use Doctrine\ORM\EntityManagerInterface;
+use DoctrineMigrations\Version20260923140100;
+use Psr\Log\NullLogger;
+use Symfony\Bundle\FrameworkBundle\Test\KernelTestCase;
+
+/**
+ * State tokens follow commit order per account (Version20260923140100).
+ *
+ * The identity value used to be the token and was drawn at INSERT, so two
+ * writers of one account could commit out of order and a client holding the
+ * later token never saw the earlier row. What proves the fix is the lock: a
+ * transaction that has written a change row for an account holds that
+ * account's lock until it commits, so nobody else can draw a number for it
+ * in between — and the number Doctrine hands back is the one stored.
+ *
+ * The migration's own statements are applied inside this test's transaction,
+ * so the test runs against a database that has not been migrated yet and
+ * leaves it exactly as it found it.
+ */
+final class ChangeLogCommitOrderTest extends KernelTestCase
+{
+    private const int LOCK_NAMESPACE = 1246576976;
+    private const int ACCOUNT        = 2_000_000_101;
+    private const int OTHER_ACCOUNT  = 2_000_000_102;
+
+    private Connection $connection;
+    private ?Connection $observer = null;
+
+    protected function setUp(): void
+    {
+        self::bootKernel();
+
+        $this->connection = self::getContainer()->get(Connection::class);
+        $this->connection->beginTransaction();
+
+        require_once \dirname(__DIR__, 3) . '/migrations/Version20260923140100.php';
+
+        $migration = new Version20260923140100($this->connection, new NullLogger());
+        $migration->up(new Schema());
+
+        foreach ($migration->getSql() as $query) {
+            $this->connection->executeStatement($query->getStatement());
+        }
+    }
+
+    protected function tearDown(): void
+    {
+        $this->observer?->close();
+
+        if (true === $this->connection->isTransactionActive()) {
+            $this->connection->rollBack();
+        }
+
+        parent::tearDown();
+    }
+
+    public function testAWriterHoldsItsAccountUntilCommitAndKeepsTheStoredNumber(): void
+    {
+        $container = self::getContainer();
+        $em        = $container->get(EntityManagerInterface::class);
+
+        $container->get(StateManager::class)->recordCreated(self::ACCOUNT, JmapObjectType::Email, 'e-1');
+        $em->flush();
+
+        $row = $em->getRepository(ChangeLog::class)->findOneBy(['accountId' => self::ACCOUNT]);
+        self::assertNotNull($row);
+
+        $stored = (int) $this->connection->fetchOne(
+            'SELECT sequence FROM jmap_change_log WHERE account_id = ?',
+            [self::ACCOUNT],
+        );
+
+        self::assertSame($stored, $row->sequence, 'the entity carries the number the trigger stored');
+
+        // A second session, as a concurrent writer would be.
+        $this->observer = DriverManager::getConnection($this->connection->getParams());
+
+        self::assertFalse(
+            (bool) $this->observer->fetchOne(
+                'SELECT pg_try_advisory_xact_lock(?, ?)',
+                [self::LOCK_NAMESPACE, self::ACCOUNT],
+            ),
+            'another writer of this account must wait for the commit',
+        );
+
+        self::assertTrue(
+            (bool) $this->observer->fetchOne(
+                'SELECT pg_try_advisory_xact_lock(?, ?)',
+                [self::LOCK_NAMESPACE, self::OTHER_ACCOUNT],
+            ),
+            'writers of other accounts are not held up',
+        );
+    }
+}