Xauele · Oct 26, 2026
diff --git a/app/Actions/CreateConversationEvent.php b/app/Actions/CreateConversationEvent.php
@@ -9,6 +9,7 @@
 use App\Models\ConversationParticipant;
 use App\Models\Message;
 use App\Models\User;
+use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Str;
 
 final readonly class CreateConversationEvent
@@ -19,22 +20,26 @@
      */
     public function handle(User $actor, Conversation $org, OrgEvent $event, ?User $subject = null, string $body = ''): Message
     {
-        $message = Message::query()->create([
-            'conversation_id' => $org->id,
-            'user_id' => $actor->id,
-            'client_id' => Str::uuid()->toString(),
-            'body' => $body,
-            'event' => $event,
-            'subject_id' => $subject?->id,
-        ]);
-
-        CreateMessage::advance($org, $message, unstash: false);
-
-        ConversationParticipant::query()
-            ->where('conversation_id', $org->id)
-            ->where('user_id', $actor->id)
-            ->update(['read_through_id' => $message->id]);
-
-        return $message;
+        return DB::transaction(function () use ($actor, $org, $event, $subject, $body): Message {
+            Conversation::query()->whereKey($org->id)->lockForUpdate()->firstOrFail();
+
+            $message = Message::query()->create([
+                'conversation_id' => $org->id,
+                'user_id' => $actor->id,
+                'client_id' => Str::uuid()->toString(),
+                'body' => $body,
+                'event' => $event,
+                'subject_id' => $subject?->id,
+            ]);
+
+            CreateMessage::advance($org, $message, unstash: false);
+
+            ConversationParticipant::query()
+                ->where('conversation_id', $org->id)
+                ->where('user_id', $actor->id)
+                ->update(['read_through_id' => $message->id]);
+
+            return $message;
+        });
     }
 }
diff --git a/app/Actions/CreateConversationPing.php b/app/Actions/CreateConversationPing.php
@@ -22,6 +22,8 @@
     public function handle(User $sender, Conversation $conversation): Message
     {
         return DB::transaction(function () use ($sender, $conversation): Message {
+            Conversation::query()->whereKey($conversation->id)->lockForUpdate()->firstOrFail();
+
             $message = Message::query()->create([
                 'conversation_id' => $conversation->id,
                 'user_id' => $sender->id,

diff --git a/app/Actions/CreateMessage.php b/app/Actions/CreateMessage.php
@@ -65,6 +65,10 @@ public function handle(User $sender, Conversation $conversation, string $body, s
         };
 
         $message = DB::transaction(function () use ($sender, $conversation, $body, $clientId, $replyTo, $images, $imageDescriptions, $url, $voice, $voiceWaveform): Message {
+            // PostgreSQL IDs follow INSERT order, not commit order. Serialize writers before
+            // INSERT so cursor polling cannot miss a lower ID committed after a higher one.
+            Conversation::query()->whereKey($conversation->id)->lockForUpdate()->firstOrFail();
+
             $linkPreview = $url === null ? null : LinkPreview::query()->firstOrCreate(['url_hash' => hash('sha256', $url)], ['url' => $url]);
 
             $message = Message::query()->createOrFirst(

diff --git a/app/Actions/CreateOrgMembers.php b/app/Actions/CreateOrgMembers.php
@@ -27,6 +27,9 @@ public function __construct(private CreateConversationEvent $createConversationE
     public function handle(User $actor, Conversation $org, Collection $members, bool $sharesHistory): void
     {
         DB::transaction(function () use ($actor, $org, $members, $sharesHistory): void {
+            // Match message writers' lock order before touching participants.
+            Conversation::query()->whereKey($org->id)->lockForUpdate()->firstOrFail();
+
             $latest = $org->fresh()->last_message_id ?? 0;
 
             foreach ($members as $member) {

diff --git a/app/Actions/DeleteOrgMember.php b/app/Actions/DeleteOrgMember.php
@@ -25,6 +25,8 @@ public function __construct(private CreateConversationEvent $createConversationE
     public function handle(User $actor, Conversation $org, User $member): void
     {
         DB::transaction(function () use ($actor, $org, $member): void {
+            Conversation::query()->whereKey($org->id)->lockForUpdate()->firstOrFail();
+
             ConversationParticipant::query()->where('conversation_id', $org->id)->where('user_id', $member->id)->delete();
 
             $remaining = $org->participants()->with('user')->oldest('id')->get();

diff --git a/app/Actions/UpdateOrgMemberRole.php b/app/Actions/UpdateOrgMemberRole.php
@@ -21,6 +21,8 @@ public function __construct(private CreateConversationEvent $createConversationE
     public function handle(User $actor, Conversation $org, User $member, OrgRole $role): void
     {
         DB::transaction(function () use ($actor, $org, $member, $role): void {
+            Conversation::query()->whereKey($org->id)->lockForUpdate()->firstOrFail();
+
             $updated = ConversationParticipant::query()
                 ->where('conversation_id', $org->id)
                 ->where('user_id', $member->id)

diff --git a/tests/Feature/RemotesCursorConcurrencyTest.php b/tests/Feature/RemotesCursorConcurrencyTest.php
@@ -0,0 +1,152 @@
+<?php
+
+declare(strict_types=1);
+
+use App\Actions\CreateConversation;
+use App\Models\User;
+use App\Queries\ConversationThreadQuery;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\File;
+
+it('publishes concurrent Remotes messages in cursor order', function (): void {
+    $database = DB::connection()->getDatabaseName();
+    $originalConnection = DB::getDefaultConnection();
+
+    // RefreshDatabase wraps the ordinary connection in a transaction. These fixtures must be
+    // committed to be visible to the two independent worker processes.
+    config()->set('database.connections.cursor_probe', [
+        ...config('database.connections.pgsql'),
+        'database' => $database,
+    ]);
+
+    DB::setDefaultConnection('cursor_probe');
+
+    try {
+        $alice = User::factory()->withRemotes()->create();
+        $bob = User::factory()->withRemotes()->create();
+        $conversation = resolve(CreateConversation::class)->handle($alice, $bob);
+    } finally {
+        DB::setDefaultConnection($originalConnection);
+    }
+
+    $directory = sys_get_temp_dir().'/devnull-cursor-'.bin2hex(random_bytes(8));
+    mkdir($directory, 0700);
+    $release = $directory.'/release';
+    $ready = $directory.'/ready';
+    $firstFinished = $directory.'/first';
+    $secondFinished = $directory.'/second';
+    $secondTag = 'devnull-cursor-'.bin2hex(random_bytes(8));
+    $processes = [];
+
+    $spawn = function (string $sender, string $mode, string $finished, string $tag) use ($database, $conversation, $ready, $release, &$processes): void {
+        $environment = [
+            ...getenv(),
+            'APP_ENV' => 'testing',
+            'APP_KEY' => (string) config('app.key'),
+            'DB_CONNECTION' => 'pgsql',
+            'DB_DATABASE' => $database,
+            'DB_HOST' => (string) config('database.connections.pgsql.host'),
+            'DB_PORT' => (string) config('database.connections.pgsql.port'),
+            'DB_USERNAME' => (string) config('database.connections.pgsql.username'),
+            'DB_PASSWORD' => (string) config('database.connections.pgsql.password'),
+            'QUEUE_CONNECTION' => 'sync',
+            'WEB_PUSH_PUBLIC_KEY' => '',
+            'WEB_PUSH_PRIVATE_KEY' => '',
+        ];
+
+        $process = proc_open(
+            [PHP_BINARY, base_path('tests/Support/RemotesCursorWorker.php'), $sender, $conversation->id, $mode, $ready, $release, $finished, $tag],
+            [0 => ['pipe', 'r'], 1 => ['pipe', 'w'], 2 => ['pipe', 'w']],
+            $pipes,
+            base_path(),
+            $environment,
+        );
+
+        throw_unless(is_resource($process), RuntimeException::class, 'Could not start the Remotes concurrency worker.');
+
+        fclose($pipes[0]);
+        $processes[] = [$process, $pipes];
+    };
+
+    $await = static function (Closure $condition, string $description): void {
+        $deadline = microtime(true) + 20;
+
+        while (! $condition()) {
+            throw_if(microtime(true) >= $deadline, RuntimeException::class, 'Timed out: '.$description);
+
+            time_nanosleep(0, 20_000_000);
+        }
+    };
+
+    try {
+        $spawn($alice->id, 'hold', $firstFinished, 'devnull-cursor-first');
+        $await(fn (): bool => is_file($ready), 'first writer did not reach its insert');
+
+        $spawn($bob->id, 'send', $secondFinished, $secondTag);
+        $outcome = null;
+
+        $await(function () use (&$outcome, $secondFinished, $secondTag): bool {
+            if (is_file($secondFinished)) {
+                $outcome = 'committed_before_first';
+            } else {
+                $waiting = DB::connection('cursor_probe')->selectOne(
+                    'select count(*) as total from pg_stat_activity where application_name = ? and wait_event_type = ?',
+                    [$secondTag, 'Lock'],
+                );
+
+                if ((int) $waiting->total > 0) {
+                    $outcome = 'waiting_for_first';
+                }
+            }
+
+            return $outcome !== null;
+        }, 'second writer neither committed nor waited on the first');
+
+        // Without a lock before INSERT, the second worker commits its higher ID while the
+        // first worker is still paused inside its transaction. A polling cursor then loses
+        // the first message forever.
+        expect($outcome)->toBe('waiting_for_first');
+
+        file_put_contents($release, 'go');
+
+        $await(fn (): bool => is_file($firstFinished) && is_file($secondFinished), 'workers did not commit');
+
+        $firstId = (int) file_get_contents($firstFinished);
+        $secondId = (int) file_get_contents($secondFinished);
+        $messages = resolve(ConversationThreadQuery::class)->get($alice, $bob)['messages']->pluck('id')->all();
+        $incremental = resolve(ConversationThreadQuery::class)->get($alice, $bob, $firstId)['messages']->pluck('id')->all();
+
+        expect($firstId)->toBeLessThan($secondId)
+            ->and($messages)->toBe([$firstId, $secondId])
+            ->and($incremental)->toBe([$secondId]);
+    } finally {
+        file_put_contents($release, 'go');
+
+        foreach ($processes as [$process, $pipes]) {
+            $deadline = microtime(true) + 5;
+
+            while (proc_get_status($process)['running'] && microtime(true) < $deadline) {
+                time_nanosleep(0, 20_000_000);
+            }
+
+            if (proc_get_status($process)['running']) {
+                proc_terminate($process);
+            }
+
+            fclose($pipes[1]);
+            fclose($pipes[2]);
+            proc_close($process);
+        }
+
+        DB::connection('cursor_probe')->table('conversations')->where('id', $conversation->id)->delete();
+        DB::connection('cursor_probe')->table('users')->whereIn('id', [$alice->id, $bob->id])->delete();
+        DB::connection('cursor_probe')->table('profiles')->whereIn('github_id', [$alice->github_id, $bob->github_id])->delete();
+        DB::purge('cursor_probe');
+
+        foreach (glob($directory.'/*') ?: [] as $file) {
+            File::delete($file);
+        }
+
+        rmdir($directory);
+    }
+});
diff --git a/tests/Support/RemotesCursorWorker.php b/tests/Support/RemotesCursorWorker.php
@@ -0,0 +1,55 @@
+<?php
+
+declare(strict_types=1);
+
+use App\Actions\CreateMessage;
+use App\Models\Conversation;
+use App\Models\User;
+use Illuminate\Contracts\Console\Kernel;
+use Illuminate\Database\Events\QueryExecuted;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Sleep;
+
+require __DIR__.'/../../vendor/autoload.php';
+
+$app = require __DIR__.'/../../bootstrap/app.php';
+$app->make(Kernel::class)->bootstrap();
+
+[$script, $senderId, $conversationId, $mode, $ready, $release, $finished, $tag] = $argv;
+
+DB::select('select set_config(?, ?, false)', ['application_name', $tag]);
+
+if ($mode === 'hold') {
+    DB::listen(static function (QueryExecuted $query) use ($ready, $release): void {
+        if (! str_contains($query->sql, 'insert into "messages"')) {
+            return;
+        }
+
+        file_put_contents($ready, 'inserted');
+        $deadline = microtime(true) + 30;
+
+        while (! is_file($release)) {
+            throw_if(microtime(true) >= $deadline, RuntimeException::class, 'Timed out waiting to release the first insert.');
+
+            Sleep::usleep(20_000);
+        }
+    });
+}
+
+try {
+    $sender = User::query()->findOrFail($senderId);
+    $conversation = Conversation::query()->findOrFail($conversationId);
+
+    $message = resolve(CreateMessage::class)->handle(
+        $sender,
+        $conversation,
+        'concurrent cursor probe',
+        $mode === 'hold' ? '96c98d0d-e42e-40aa-b7d2-42490573a016' : '96c98d0d-e42e-40aa-b7d2-42490573a017',
+        linkPreview: false,
+    );
+
+    file_put_contents($finished, (string) $message->id);
+} catch (Throwable $throwable) {
+    fwrite(STDERR, $throwable->__toString().PHP_EOL);
+    exit(1);
+}