Repository: laravel/reverb (php)

Issue:
Potential Issue with Pusher Ping/Pong Messages
### Reverb Version

1.0.0-beta14

### Laravel Version

10.10

### PHP Version

8.1

### Description

If I am understanding the Pusher protocol correctly, then if the WebSocket implementation supports native ping/pong messages, then Pusher will use those, and the `pusher:ping` and `pusher:pong` messages are only used if the client doesn't support websocket protocol ping messages.
[Link to Pusher docs here](https://pusher.com/docs/channels/library_auth_reference/pusher-websockets-protocol/#ping-and-pong-messages)

It seems like Laravel Reverb is ONLY sending `pusher:ping` and not using the websocket native ping messages to determine if a connection is still alive. I found this to be a compatibility issue with the [official dotnet Pusher client](https://github.com/pusher/pusher-websocket-dotnet). If I understand the situation correctly, because the C# implementation of the websocket client handles ping messages at the ws protocol level, they didn't implement any handlers for `pusher:ping` messages, which makes this client incompatible with Reverb.

If my assumptions are correct then this means Reverb should detect the client version and use a different pinging strategy based on the version? Or in my case could I just disable the reverb ping messages entirely and rely on the websocket pings being handled internally to keep the connection alive?

**Edit:** This excerpt from the Pusher docs linked above seems to imply that the C# client's behaviour is fine:

> If the WebSocket draft supports protocol level ping-pong, then on receipt of a ping message, the client MUST respond with a pong message.
> 
> If the client does not support protocol level pings and advertises (on connect) that it implements a protocol version >= 5 then the client MUST respond to a pusher:ping event with a pusher.pong event.

It doesn't explicitly state that clients that already support protocol level pings must ALSO respond to pusher:ping events.

### Steps To Reproduce

Set up a minimal Reverb server and minimal Pusher dotnet client (https://github.com/pusher/pusher-websocket-dotnet), wait 1 minute after connection, server drops client connection due to missed ping and client reconnects.


Discussion:
So websocket ping/pong is handled here: https://github.com/laravel/reverb/blob/838387607b9223b67eee592a916b36900374bb14/src/Servers/Reverb/Connection.php#L84

Maybe this should call `touch()` on the Pusher connection to reset `lastSeenAt`, and then the `pusher:ping` fallback won't be necessary? 

I agree with above approach, probably receiving any control frame should `touch()` the connection the same way any non-control frame does? But to fully meet the spec, sending a `ping()` or `pong()` should also attempt to send the control frame op-code as well?

Lastly, while playing around to check this, I found that `PingInactiveConnections` occurs on a fixed 60 second interval on the `$loop` but my hunch is that this should respect the configured `ping_interval` value? Otherwise it seems possible that active connections may be incorrectly treated at inactive.

https://github.com/laravel/reverb/blob/838387607b9223b67eee592a916b36900374bb14/src/Servers/Reverb/Console/Commands/StartServer.php#L91-L97


I don't think calling `touch()` on receipt of the control frame fully solves the issue since the dotnet client doesn't respond to `pusher:ping` so Reverb will never see a response from the client when pinging inactive connections and ultimately terminate the connection.

I'll take a look to see what's involved in adding support for this.

I notice there are some other similar issues on the repo so wonder if this should be addressed by the client. Doesn't look, at first glance, that there are any checks for the advertised supported pusher protocols.


Yes but by calling `touch()` in response to a control frame event would update the connection's `lastSeenAt` and it wouldn't be included in the next round of `PingInactiveConnections` anyway right? So this _should_ work as long as the client is indeed sending ping events reliably.

What do you think?
> I notice there are some other similar issues on the repo so wonder if this should be addressed by the client. Doesn't look, at first glance, that there are any checks for the advertised supported pusher protocols.

I believe that because the dotnet WebSocket client supports protocol level pings they didn't bother to include support for `pusher:ping` messages, since the spec states you can use either method for detecting inactive connections. This is likely an issue with more client implementations out there. You don't need to check the advertised pusher version to know if ping is supported, you check the `Sec-Websocket-Version` header on conection (I think)

We're seeing the same issue using the https://github.com/pusher/pusher-websocket-swift package where (unlike the Echo / Pusher JS client), it is not sending `pusher:ping` messages and therefore the connections are pruned unexpectedly
Appears most non web based clients don't use `pusher:ping` so most (all?) mobile clients don't work with Reverb.

Patch:
diff --git a/src/Connection.php b/src/Connection.php
--- a/src/Connection.php
+++ b/src/Connection.php
@@ -5,6 +5,7 @@
 use Laravel\Reverb\Concerns\GeneratesIdentifiers;
 use Laravel\Reverb\Contracts\Connection as ConnectionContract;
 use Laravel\Reverb\Events\MessageSent;
+use Ratchet\RFC6455\Messaging\Frame;
 
 class Connection extends ConnectionContract
 {
@@ -50,6 +51,14 @@ public function send(string $message): void
         MessageSent::dispatch($this, $message);
     }
 
+    /**
+     * Send a control frame to the connection.
+     */
+    public function control(string $type = Frame::OP_PING): void
+    {
+        $this->connection->send(new Frame('', opcode: $type));
+    }
+
     /**
      * Terminate a connection.
      */
diff --git a/src/Contracts/Connection.php b/src/Contracts/Connection.php
--- a/src/Contracts/Connection.php
+++ b/src/Contracts/Connection.php
@@ -3,6 +3,7 @@
 namespace Laravel\Reverb\Contracts;
 
 use Laravel\Reverb\Application;
+use Ratchet\RFC6455\Messaging\Frame;
 
 abstract class Connection
 {
@@ -16,6 +17,11 @@ abstract class Connection
      */
     protected $hasBeenPinged = false;
 
+    /**
+     * Indicates if the connection uses control frames.
+     */
+    protected $usesControlFrames = false;
+
     /**
      * Create a new connection instance.
      */
@@ -39,6 +45,11 @@ abstract public function id(): string;
      */
     abstract public function send(string $message): void;
 
+    /**
+     * Send a control frame to the connection.
+     */
+    abstract public function control(string $type = Frame::OP_PING): void;
+
     /**
      * Terminate a connection.
      */
@@ -136,4 +147,22 @@ public function isStale(): bool
     {
         return $this->isInactive() && $this->hasBeenPinged;
     }
+
+    /**
+     * Determine whether the connection uses control frames.
+     */
+    public function usesControlFrames(): bool
+    {
+        return $this->usesControlFrames;
+    }
+
+    /**
+     * Mark the connection as using control frames to track activity.
+     */
+    public function setUsesControlFrames(bool $usesControlFrames = true): Connection
+    {
+        $this->usesControlFrames = $usesControlFrames;
+
+        return $this;
+    }
 }
diff --git a/src/Protocols/Pusher/EventHandler.php b/src/Protocols/Pusher/EventHandler.php
--- a/src/Protocols/Pusher/EventHandler.php
+++ b/src/Protocols/Pusher/EventHandler.php
@@ -117,7 +117,9 @@ public function pong(Connection $connection): void
      */
     public function ping(Connection $connection): void
     {
-        static::send($connection, 'ping');
+        $connection->usesControlFrames() 
+            ? $connection->control() 
+            : static::send($connection, 'ping');
 
         $connection->ping();
     }
diff --git a/src/Protocols/Pusher/Http/Controllers/PusherController.php b/src/Protocols/Pusher/Http/Controllers/PusherController.php
--- a/src/Protocols/Pusher/Http/Controllers/PusherController.php
+++ b/src/Protocols/Pusher/Http/Controllers/PusherController.php
@@ -8,6 +8,7 @@
 use Laravel\Reverb\Protocols\Pusher\Server as PusherServer;
 use Laravel\Reverb\Servers\Reverb\Connection;
 use Psr\Http\Message\RequestInterface;
+use Ratchet\RFC6455\Messaging\FrameInterface;
 
 class PusherController
 {
@@ -34,6 +35,10 @@ public function __invoke(RequestInterface $request, Connection $connection, stri
             fn ($message) => $this->server->message($reverbConnection, (string) $message)
         );
 
+        $connection->onControl(
+            fn (FrameInterface $message) => $this->server->control($reverbConnection, $message)
+        );
+
         $connection->onClose(
             fn () => $this->server->close($reverbConnection)
         );
diff --git a/src/Protocols/Pusher/Server.php b/src/Protocols/Pusher/Server.php
--- a/src/Protocols/Pusher/Server.php
+++ b/src/Protocols/Pusher/Server.php
@@ -10,6 +10,8 @@
 use Laravel\Reverb\Protocols\Pusher\Contracts\ChannelManager;
 use Laravel\Reverb\Protocols\Pusher\Exceptions\InvalidOrigin;
 use Laravel\Reverb\Protocols\Pusher\Exceptions\PusherException;
+use Ratchet\RFC6455\Messaging\Frame;
+use Ratchet\RFC6455\Messaging\FrameInterface;
 
 class Server
 {
@@ -70,6 +72,21 @@ public function message(Connection $from, string $message): void
         }
     }
 
+    /**
+     * Handle a low-level WebSocket control frame.
+     */
+    public function control(Connection $from, FrameInterface $message): void
+    {
+        Log::info('Control Frame Received', $from->id());
+        Log::message($message);
+
+        $from->setUsesControlFrames();
+
+        if (in_array($message->getOpcode(), [Frame::OP_PING, Frame::OP_PONG], strict: true)) {
+            $from->touch();
+        }
+    }
+
     /**
      * Handle a client disconnection.
      */
diff --git a/src/Servers/Reverb/Connection.php b/src/Servers/Reverb/Connection.php
--- a/src/Servers/Reverb/Connection.php
+++ b/src/Servers/Reverb/Connection.php
@@ -27,6 +27,13 @@ class Connection extends EventEmitter implements WebSocketConnection
      */
     protected $onMessage;
 
+    /**
+     * The control frame handler.
+     *
+     * @var ?callable
+     */
+    protected $onControl;
+
     /**
      * The connection close handler.
      *
@@ -83,6 +90,10 @@ public function send(mixed $message): void
      */
     public function control(FrameInterface $message): void
     {
+        if ($this->onControl) {
+            ($this->onControl)($message);
+        }
+
         match ($message->getOpcode()) {
             Frame::OP_PING => $this->send(new Frame($message->getPayload(), opcode: Frame::OP_PONG)),
             Frame::OP_PONG => fn () => null,
@@ -98,6 +109,14 @@ public function onMessage(callable $callback): void
         $this->onMessage = $callback;
     }
 
+    /**
+     * Set the control frame handler.
+     */
+    public function onControl(callable $callback): void
+    {
+        $this->onControl = $callback;
+    }
+
     /**
      * Set the close handler.
      */


Test patch:
diff --git a/tests/FakeConnection.php b/tests/FakeConnection.php
--- a/tests/FakeConnection.php
+++ b/tests/FakeConnection.php
@@ -7,6 +7,7 @@
 use Laravel\Reverb\Concerns\GeneratesIdentifiers;
 use Laravel\Reverb\Contracts\ApplicationProvider;
 use Laravel\Reverb\Contracts\Connection as BaseConnection;
+use Ratchet\RFC6455\Messaging\Frame;
 
 class FakeConnection extends BaseConnection
 {
@@ -97,6 +98,12 @@ public function send(string $message): void
         $this->messages[] = $message;
     }
 
+    /**
+     * Send a control frame to the connection.
+     */
+    public function control(string $type = Frame::OP_PING): void { }
+
+
     /**
      * Terminate a connection.
      */
diff --git a/tests/Feature/Protocols/Pusher/Reverb/ServerTest.php b/tests/Feature/Protocols/Pusher/Reverb/ServerTest.php
--- a/tests/Feature/Protocols/Pusher/Reverb/ServerTest.php
+++ b/tests/Feature/Protocols/Pusher/Reverb/ServerTest.php
@@ -4,6 +4,7 @@
 use Laravel\Reverb\Jobs\PingInactiveConnections;
 use Laravel\Reverb\Jobs\PruneStaleConnections;
 use Laravel\Reverb\Tests\ReverbTestCase;
+use Ratchet\RFC6455\Messaging\Frame;
 use React\Promise\Deferred;
 
 use function Ratchet\Client\connect as wsConnect;
@@ -482,3 +483,57 @@
     expect($response)->toContain('"ids\":[1]');
     expect($response)->toContain('"hash\":{\"1\":{\"name\":\"Test User\"}}');
 });
+
+it('can handle a ping control frame', function () {
+    $connection = connect();
+    subscribe('test-channel', connection: $connection);
+    $channels = channels();
+    $managedConnection = Arr::first($channels->connections());
+    $subscribedAt = $managedConnection->lastSeenAt();
+    sleep(1);
+    $connection->send(new Frame('', opcode: Frame::OP_PING));
+    
+    $connection->assertPonged();
+    expect($managedConnection->lastSeenAt())->toBeGreaterThan($subscribedAt);
+});
+
+it('can handle a pong control frame', function () {
+    $connection = connect();
+    subscribe('test-channel', connection: $connection);
+    $channels = channels();
+    $managedConnection = Arr::first($channels->connections());
+    $subscribedAt = $managedConnection->lastSeenAt();
+    sleep(1);
+    $connection->send(new Frame('', opcode: Frame::OP_PONG));
+    
+    $connection->assertNotPinged();
+    $connection->assertNotPonged();
+    expect($managedConnection->lastSeenAt())->toBeGreaterThan($subscribedAt);
+});
+
+it('uses pusher control messages by default', function () {
+    $connection = connect();
+    subscribe('test-channel', connection: $connection);
+
+    $channels = channels();
+    Arr::first($channels->connections())->setLastSeenAt(time() - 60 * 10);
+
+    (new PingInactiveConnections)->handle($channels);
+
+    $connection->assertReceived('{"event":"pusher:ping"}');
+    $connection->assertNotPinged();
+});
+
+it('uses control frames when the client prefers', function () {
+    $connection = connect();
+    $connection->send(new Frame('', opcode: Frame::OP_PING));
+    subscribe('test-channel', connection: $connection);
+
+    $channels = channels();
+    Arr::first($channels->connections())->setLastSeenAt(time() - 60 * 10);
+
+    (new PingInactiveConnections)->handle($channels);
+
+    $connection->assertPinged();
+    $connection->assertNotReceived('{"event":"pusher:ping"}');
+});
diff --git a/tests/TestConnection.php b/tests/TestConnection.php
--- a/tests/TestConnection.php
+++ b/tests/TestConnection.php
@@ -25,6 +25,10 @@ class TestConnection
      */
     public $receivedMessages = [];
 
+    public $wasPinged = false;
+
+    public $wasPonged = false;
+
     /**
      * Create a new test connection instance.
      */
@@ -34,6 +38,14 @@ public function __construct(public WebSocket $connection)
             $this->receivedMessages[] = (string) $message;
         });
 
+        $connection->on('ping', function () {
+            $this->wasPinged = true;
+        });
+
+        $connection->on('pong', function () {
+            $this->wasPonged = true;
+        });
+
         $connection->on('close', function ($code, $message) {
             $this->receivedMessages[] = (string) $message;
         });
@@ -89,6 +101,58 @@ public function assertReceived(string $message, ?int $count = null): void
         expect($this->receivedMessages)->toContain($message);
     }
 
+    /**
+     * Assert that the connection did not receiv the given message.
+     */
+    public function assertNotReceived(string $message): void
+    {
+        if (! in_array($message, $this->receivedMessages)) {
+            $this->await();
+        }
+
+        expect($this->receivedMessages)->not->toContain($message);
+    }
+
+    /**
+     * Assert the connection was pinged during the test.
+     */
+    public function assertPinged(): void
+    {
+        $this->await();
+
+        expect($this->wasPinged)->toBeTrue();
+    }
+
+    /**
+     * Assert the connection was not pinged during the test.
+     */
+    public function assertNotPinged(): void
+    {
+        $this->await();
+
+        expect($this->wasPinged)->toBeFalse();
+    }
+
+    /**
+     * Assert the connection was ponged during the test.
+     */
+    public function assertPonged(): void
+    {
+        $this->await();
+
+        expect($this->wasPonged)->toBeTrue();
+    }
+
+    /**
+     * Assert the connection was not ponged during the test.
+     */
+    public function assertNotPonged(): void
+    {
+        $this->await();
+
+        expect($this->wasPonged)->toBeFalse();
+    }
+
     /**
      * Proxy method calls to the connection.
      */