Add optional keepalive and session timeouts (#23)

cyppe · web-flow · commit 0dab4334e181 · 2026-02-15T14:47:00.000+08:00
diff --git a/README-CN.md b/README-CN.md
@@ -102,6 +102,22 @@ TRIGGER_PASSWORD=password
 ...
 ~~~
 
+### 连接超时（建议）
+
+如果你的 MySQL（或其前置代理）会断开长时间空闲的连接（例如 `wait_timeout` / `interactive_timeout` 设置较小），trigger 进程可能会出现类似错误并退出：
+
+`SQLSTATE[HY000] ... 4031 The client was disconnected by the server because of inactivity.`
+
+为提升稳定性，你可以开启 keepalive 心跳查询，及/或为 trigger 的 MySQL 元数据连接设置 session 变量：
+
+~~~env
+# 定期 ping MySQL（秒）。设置为 0 表示禁用。
+TRIGGER_KEEPALIVE=60
+
+# 在连接建立后应用的 session 变量（逗号分隔 key=value）。
+TRIGGER_SESSION_VARIABLES=wait_timeout=7200,interactive_timeout=7200,net_read_timeout=3600,net_write_timeout=3600
+~~~
+
 ## 启动服务
 
 ~~~bash
diff --git a/README.md b/README.md
@@ -106,6 +106,22 @@ TRIGGER_PASSWORD=password
 ...
 ~~~
 
+### Connection Timeouts (Recommended)
+
+If your MySQL server (or a proxy in front of it) disconnects idle clients (low `wait_timeout` / `interactive_timeout`), the trigger process may crash with errors like:
+
+`SQLSTATE[HY000] ... 4031 The client was disconnected by the server because of inactivity.`
+
+To improve stability, enable a keepalive ping and/or set session variables for the trigger's MySQL metadata connection:
+
+~~~env
+# Ping MySQL periodically (seconds). Set to 0 to disable.
+TRIGGER_KEEPALIVE=60
+
+# Session variables applied on connect (comma-separated key=value pairs).
+TRIGGER_SESSION_VARIABLES=wait_timeout=7200,interactive_timeout=7200,net_read_timeout=3600,net_write_timeout=3600
+~~~
+
 ## Usage
 
 Start the trigger service to begin listening for MySQL events:
diff --git a/config/trigger.php b/config/trigger.php
@@ -25,6 +25,17 @@
             'tables' => env('TRIGGER_TABLES', '') ? explode(',', env('TRIGGER_TABLES')) : [],
 
             'heartbeat' => (int) env('TRIGGER_HEARTBEAT', 3),
+
+            // Periodically ping the MySQL metadata connection to avoid server-side idle disconnects.
+            // Set to 0 to disable.
+            'keepalive' => (int) env('TRIGGER_KEEPALIVE', 0),
+
+            // MySQL session variables to apply on connect (for the metadata connection).
+            // Example:
+            // - wait_timeout=7200,interactive_timeout=7200
+            'session_variables' => env('TRIGGER_SESSION_VARIABLES', '')
+                ? array_filter(array_map('trim', explode(',', (string) env('TRIGGER_SESSION_VARIABLES'))))
+                : [],
             'subscribers' => [
                 // Huangdijia\Trigger\Subscribers\Heartbeat::class,
             ],
diff --git a/src/Console/StartCommand.php b/src/Console/StartCommand.php
@@ -11,9 +11,12 @@
 
 namespace Huangdijia\Trigger\Console;
 
+use Doctrine\DBAL\Exception as DbalException;
 use Huangdijia\Trigger\Facades\Trigger;
 use Illuminate\Console\Command;
 use MySQLReplication\Exception\MySQLReplicationException;
+use PDOException;
+use Throwable;
 
 class StartCommand extends Command
 {
@@ -91,6 +94,18 @@ public function handle(): int
             $this->info('Retry now');
             sleep(1);
 
+            goto start;
+        } catch (DbalException|PDOException $e) {
+            $this->error($e->getMessage());
+
+            if (! $this->shouldRetry($e)) {
+                throw $e;
+            }
+
+            // Keep current binlog position so we can resume after reconnect.
+            $this->info('Retry now');
+            sleep(1);
+
             goto start;
         }
 
@@ -123,4 +138,32 @@ protected function listenForSignals(): void
         pcntl_signal(SIGTERM, $handler);
         pcntl_signal(SIGINT, $handler);
     }
+
+    private function shouldRetry(Throwable $e): bool
+    {
+        $pdo = $e instanceof PDOException ? $e : null;
+
+        if ($pdo === null && $e->getPrevious() instanceof PDOException) {
+            $pdo = $e->getPrevious();
+        }
+
+        if ($pdo !== null && isset($pdo->errorInfo[1])) {
+            $driverCode = (int) $pdo->errorInfo[1];
+
+            return in_array($driverCode, [2006, 2013, 2055, 4031], true);
+        }
+
+        $message = strtolower($e->getMessage());
+
+        if (str_contains($message, 'access denied') || str_contains($message, 'unknown database')) {
+            return false;
+        }
+
+        return str_contains($message, 'server has gone away')
+            || str_contains($message, 'lost connection')
+            || str_contains($message, 'disconnected by the server')
+            || str_contains($message, 'connection refused')
+            || str_contains($message, 'connection timed out')
+            || str_contains($message, 'broken pipe');
+    }
 }
diff --git a/src/Trigger.php b/src/Trigger.php
@@ -12,6 +12,7 @@
 namespace Huangdijia\Trigger;
 
 use Closure;
+use Doctrine\DBAL\Connection;
 use Exception;
 use Illuminate\Container\Container;
 use Illuminate\Contracts\Queue\ShouldQueue;
@@ -31,6 +32,10 @@ class Trigger
 {
     protected \Illuminate\Contracts\Cache\Repository $cache;
 
+    protected ?Connection $dbConnection = null;
+
+    protected int $lastKeepalivePingAt = 0;
+
     protected array $events = [];
 
     protected int $bootTime;
@@ -137,12 +142,17 @@ public function loadRoutes(): void
      */
     public function start(bool $keepUp = true): void
     {
-        tap(new MySQLReplicationFactory($this->configure($keepUp)), function (MySQLReplicationFactory $binLogStream) {
-            collect($this->getSubscribers())
-                ->reject(fn ($subscriber) => ! is_subclass_of($subscriber, EventSubscriber::class))
-                ->unique()
-                ->each(fn ($subscriber) => $binLogStream->registerSubscriber(new $subscriber($this)));
-        })->run();
+        $binLogStream = new MySQLReplicationFactory($this->configure($keepUp));
+
+        $this->dbConnection = $binLogStream->getDbConnection();
+        $this->applySessionVariables($this->dbConnection);
+
+        collect($this->getSubscribers())
+            ->reject(fn ($subscriber) => ! is_subclass_of($subscriber, EventSubscriber::class))
+            ->unique()
+            ->each(fn ($subscriber) => $binLogStream->registerSubscriber(new $subscriber($this)));
+
+        $binLogStream->run();
     }
 
     /**
@@ -183,6 +193,8 @@ public function isTerminated(): bool
     public function heartbeat(EventDTO $event): void
     {
         $this->rememberCurrent($event->getEventInfo()->binLogCurrent);
+
+        $this->keepalive();
     }
 
     /**
@@ -344,6 +356,85 @@ public function getTables(): array
         return $tables;
     }
 
+    private function applySessionVariables(?Connection $connection): void
+    {
+        if ($connection === null) {
+            return;
+        }
+
+        $variables = $this->getConfig('session_variables', []);
+
+        if (is_string($variables)) {
+            $variables = array_filter(array_map('trim', explode(',', $variables)));
+        }
+
+        if (! is_array($variables) || $variables === []) {
+            return;
+        }
+
+        foreach ($variables as $name => $value) {
+            if (is_int($name)) {
+                $pair = trim((string) $value);
+
+                if ($pair === '' || ! str_contains($pair, '=')) {
+                    continue;
+                }
+
+                [$name, $value] = array_map('trim', explode('=', $pair, 2));
+            }
+
+            if (! is_string($name) || $name === '' || ! preg_match('/^[A-Za-z0-9_]+$/', $name)) {
+                continue;
+            }
+
+            if (is_array($value) || is_object($value)) {
+                continue;
+            }
+
+            if (is_string($value) && preg_match('/^-?\d+$/', $value)) {
+                $value = (int) $value;
+            }
+
+            try {
+                $connection->executeStatement("SET SESSION {$name} = ?", [$value]);
+            } catch (Throwable) {
+                // Ignore session variable failures (permission/unsupported variables).
+            }
+        }
+    }
+
+    private function keepalive(): void
+    {
+        $period = (int) $this->getConfig('keepalive', 0);
+
+        if ($period <= 0 || $this->dbConnection === null) {
+            return;
+        }
+
+        $now = time();
+
+        if ($this->lastKeepalivePingAt > 0 && ($now - $this->lastKeepalivePingAt) < $period) {
+            return;
+        }
+
+        $this->lastKeepalivePingAt = $now;
+
+        try {
+            $this->dbConnection->executeQuery($this->dbConnection->getDatabasePlatform()->getDummySelectSQL());
+        } catch (Throwable) {
+            try {
+                $this->dbConnection->close();
+
+                // DBAL will reconnect automatically on the next query.
+                $this->dbConnection->executeQuery($this->dbConnection->getDatabasePlatform()->getDummySelectSQL());
+
+                $this->applySessionVariables($this->dbConnection);
+            } catch (Throwable) {
+                // If reconnect fails, the next metadata query will throw and the daemon can restart.
+            }
+        }
+    }
+
     /**
      * Parse action.
      *
