diff --git a/app/Actions/CreateConversationEvent.php b/app/Actions/CreateConversationEvent.php index 2c610e6fa..6b1c5e51f 100644 --- 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 index f9c2bbaa3..40f608e84 100644 --- 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 index d45b63fa9..f5d7774c5 100644 --- 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 index 78f424625..7070ebf30 100644 --- 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 index d49e9cab2..abf833c8a 100644 --- 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 index f70df9d3d..8d31c613d 100644 --- 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 new file mode 100644 index 000000000..5d9d56251 --- /dev/null +++ b/tests/Feature/RemotesCursorConcurrencyTest.php @@ -0,0 +1,152 @@ +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 new file mode 100644 index 000000000..c5057c5ff --- /dev/null +++ b/tests/Support/RemotesCursorWorker.php @@ -0,0 +1,55 @@ +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); +}