diff --git a/CHANGELOG.md b/CHANGELOG.md index 6fd82e6..27ba015 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,13 @@ - Added `Usage::addSamples()`, `Usage::getSampleWatermark()` and `Usage::findSamples()`. Existing events, gauges and daily rollups are unchanged and are not used as a canonical sample source. +- ClickHouse reads now carry a `max_execution_time` cap and a `query_id`, and a + read the transport abandons is reaped with `KILL QUERY`. Previously a client + socket timeout left the query running to completion and retries stacked more + of them onto the cluster. Configurable through the new `readTimeout` + constructor argument (default 25s, `null` disables); the write path is + deliberately left uncapped, and a server that refuses the settings degrades to + an uncapped read. ### Breaking diff --git a/README.md b/README.md index cd23928..403742a 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,28 @@ $usage = new Usage($adapter); $usage->setup(); // Creates events, gauges, and daily MV tables ``` +#### Read timeouts + +Reads run under a ClickHouse-side `max_execution_time` (`readTimeout`, 25s by +default) and carry a `query_id`, so a read the client gives up on is aborted by +the server and, failing that, reaped with `KILL QUERY`. Keep the cap **below** +the injected client's socket timeout — 30s on both bundled `utopia-php/client` +adapters — or the socket dies first and the query runs to completion anyway, +with retries stacking more of them onto the cluster. + +```php +$adapter = new ClickHouse( + host: 'clickhouse-server', + client: new Client((new CurlAdapter())->withTimeout(10.0)), + readTimeout: 8, // must stay under the client's 10s socket timeout +); +``` + +Writes are never capped — starving ingest is worse than a slow read. Pass +`readTimeout: null` for a `readonly = 1` ClickHouse user, which is not allowed +to change settings; the adapter also detects that rejection at runtime and +falls back to uncapped reads. + ### Using Database Adapter ```php diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index 59316c6..efc773b 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -11,6 +11,8 @@ use Throwable; use Utopia\Client; use Utopia\Client\Adapter\Curl\Client as CurlAdapter; +use Utopia\Client\Exception\ConnectionException; +use Utopia\Client\Exception\DnsException; use Utopia\Psr7\Method as HttpMethod; use Utopia\Psr7\Request\Factory as RequestFactory; use Utopia\Query\Builder\ClickHouse as ClickHouseBuilder; @@ -65,6 +67,17 @@ class ClickHouse extends SQL private const ROUTE_LOG_MAX = 1_000; + /** + * Server-side wall-clock cap for reads, in seconds. Must stay below the + * transport's socket timeout (30s for both bundled utopia-php/client + * adapters): when the socket gives up first the query still runs to + * completion, and retries stack more of them onto the cluster. + */ + private const DEFAULT_READ_TIMEOUT = 25; + + /** ClickHouse error codes returned when a user may not change settings. */ + private const SETTINGS_REJECTED_CODES = [115, 164]; + /** @var array Maps interval strings to ClickHouse time functions */ private const INTERVAL_FUNCTIONS = [ '1h' => 'toStartOfHour', @@ -135,6 +148,15 @@ class ClickHouse extends SQL */ private ?string $nextQueryId = null; + /** @var int|null Server-side execution cap for reads in seconds; null disables it */ + private readonly ?int $readTimeout; + + /** + * Cleared for the instance's remaining life once the server rejects the read + * settings, so a restricted user degrades to uncapped reads. + */ + private bool $readSettingsSupported = true; + /** * Structured log entries recorded for each routing decision. Ops * dashboards read these to confirm rollup hit-rate. @@ -183,6 +205,11 @@ class ClickHouse extends SQL * events_daily tables. When set, setup() applies a TTL that drops rows * older than the window; gauges are left untouched. Null disables TTL * (default). Must be positive. + * @param int|null $readTimeout Seconds a read may run server-side before + * ClickHouse aborts it; the abandoned query is then killed by query_id. + * Keep it below the injected client's socket timeout. Null disables both + * the cap and the cancellation — use it for a `readonly = 1` user, which + * is not allowed to change settings. Must be positive. */ public function __construct( string $host, @@ -197,13 +224,17 @@ public function __construct( bool $asyncInserts = false, bool $asyncInsertWait = true, float $dualReadSampleRate = 0.0, - ?int $retention = null + ?int $retention = null, + ?int $readTimeout = self::DEFAULT_READ_TIMEOUT ) { $this->validateHost($host); $this->validatePort($port); if ($retention !== null && $retention < 1) { throw new Exception('Retention must be a positive number of days'); } + if ($readTimeout !== null && $readTimeout < 1) { + throw new Exception('Read timeout must be a positive number of seconds'); + } if (!empty($namespace)) { $this->validateIdentifier($namespace, 'Namespace'); } @@ -223,6 +254,7 @@ public function __construct( // over-trigger the parity sampler. $this->dualReadSampleRate = max(0.0, min(1.0, $dualReadSampleRate)); $this->retention = $retention; + $this->readTimeout = $readTimeout; // `withConnectionReuse()` keeps the underlying cURL handle alive across // requests so the TCP/TLS handshake is paid once. Auth and database are @@ -514,13 +546,14 @@ private function buildErrorMessage(string $baseMessage, ?string $table = null, ? * * @param string $sql * @param array $params + * @param array $settings Per-query ClickHouse settings + * @param string|null $queryId Explicit query_id; falls back to the pinned one * @return string * @throws Exception */ - private function query(string $sql, array $params = []): string + private function query(string $sql, array $params = [], array $settings = [], ?string $queryId = null): string { - $queryId = $this->nextQueryId; - $this->nextQueryId = null; + $queryId ??= $this->consumeNextQueryId(); $scheme = $this->secure ? 'https' : 'http'; @@ -530,14 +563,18 @@ private function query(string $sql, array $params = []): string // string — avoids request-line length limits (HTTP 414) on large // `equal`/tag filters. ClickHouse does NOT parse // application/x-www-form-urlencoded bodies, so multipart is required. - // Only the tiny query_id, which has no size concern, stays in the URL. + // Only the tiny query_id and settings, which have no size concern, + // stay in the URL. $parts = ['query' => $sql]; foreach ($params as $key => $value) { $parts['param_' . $key] = $this->formatParamValue($value); } $url = "{$scheme}://{$this->host}:{$this->port}/"; if ($queryId !== null) { - $url .= '?' . http_build_query(['query_id' => $queryId]); + $settings['query_id'] = $queryId; + } + if (!empty($settings)) { + $url .= '?' . http_build_query($settings); } $this->requestCount++; @@ -562,6 +599,106 @@ private function query(string $sql, array $params = []): string return $bodyStr; } + /** + * Execute a read under a server-side execution cap, tagged with a query_id + * so an abandoned read can be reaped. Writes deliberately do not go through + * here: starving ingest is worse than a slow read. + * + * @param array $params + * @throws Exception + */ + private function queryRead(string $sql, array $params = []): string + { + if ($this->readTimeout === null || !$this->readSettingsSupported) { + return $this->query($sql, $params); + } + + // Take a pinned id — benchmarks and routing tests correlate one against + // system.query_log — and thread it through explicitly from here on. The + // shared slot must not hold an id across a request, or a concurrent + // coroutine could adopt it and be cancelled in this read's place. + $queryId = $this->consumeNextQueryId() ?? bin2hex(random_bytes(16)); + + try { + return $this->query($sql, $params, [ + 'max_execution_time' => (string) $this->readTimeout, + // A profile defaulting to `break` would return a silently + // truncated result set; usage totals must fail loudly instead. + 'timeout_overflow_mode' => 'throw', + ], $queryId); + } catch (Exception $e) { + if ($this->isSettingsRejection($e)) { + $this->readSettingsSupported = false; + return $this->query($sql, $params, [], $queryId); + } + + if (!$this->reachedServer($e)) { + throw $e; + } + + $this->killQuery($queryId); + throw $e; + } + } + + /** + * Take the caller's pinned id, if any. Read and clear in one step so a + * concurrent coroutine can't adopt an id that is already spoken for. + */ + private function consumeNextQueryId(): ?string + { + $queryId = $this->nextQueryId; + $this->nextQueryId = null; + + return $queryId; + } + + /** + * Whether the failed request got as far as the server, and so may have left + * a query running. When the connection never opened there is nothing to + * cancel, and a KILL would only burn a second socket timeout against a host + * that is already down. + */ + private function reachedServer(Exception $e): bool + { + $cause = $e->getPrevious(); + + return !($cause instanceof ConnectionException || $cause instanceof DnsException); + } + + /** + * Reap a read the transport abandoned. ASYNC so it returns without waiting + * for the query to stop, and failures are swallowed so a missing KILL + * privilege can't mask the error the caller is already surfacing. + */ + private function killQuery(string $queryId): void + { + try { + // Its own id, so the KILL never consumes one pinned for another read. + $this->query( + 'KILL QUERY WHERE query_id = {queryId:String} ASYNC', + ['queryId' => $queryId], + queryId: bin2hex(random_bytes(16)), + ); + } catch (Throwable) { + } + } + + /** + * Whether ClickHouse refused the query because the user may not change + * settings (`readonly = 1`) or does not know them (older server). + */ + private function isSettingsRejection(Exception $e): bool + { + foreach (self::SETTINGS_REJECTED_CODES as $code) { + if (str_contains($e->getMessage(), "Code: {$code}.")) { + return true; + } + } + + return false; + } + /** * Decode a ClickHouse `FORMAT JSON` response body into its data rows. * Returns an empty list when the body is not the expected envelope. @@ -1844,7 +1981,7 @@ public function getSampleWatermark(SampleRange $range, int $limit): SampleWaterm FORMAT JSON SQL; - $rows = $this->decodeRows($this->query($sql, [ + $rows = $this->decodeRows($this->queryRead($sql, [ 'environment' => $range->environment, 'region' => $range->region, 'projectInternalId' => $range->projectInternalId, @@ -1932,7 +2069,7 @@ public function findSamples(SampleRange $range, SampleWatermark $watermark, int FORMAT JSON SQL; - $rows = $this->decodeRows($this->query($sql, [ + $rows = $this->decodeRows($this->queryRead($sql, [ 'environment' => $range->environment, 'region' => $range->region, 'projectInternalId' => $range->projectInternalId, @@ -2310,7 +2447,7 @@ private function findFromTable(?string $tenant, array $queries, string $type): a $statement = $builder->build(); $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - $result = $this->query($sql, array_merge($statement->namedBindings ?? [], $extraBindings)); + $result = $this->queryRead($sql, array_merge($statement->namedBindings ?? [], $extraBindings)); $rows = $this->parseResults($result, $type); @@ -2417,7 +2554,7 @@ private function findAggregatedFromTable(?string $tenant, array $parsed, string $statement = $builder->build(); $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - $result = $this->query($sql, $statement->namedBindings ?? []); + $result = $this->queryRead($sql, $statement->namedBindings ?? []); return $this->parseAggregatedResults($result, $type); } @@ -2549,7 +2686,7 @@ private function countFromTable(string $tenant, array $queries, string $type, ?i $innerSql = $this->qualifyDdl($innerStatement->query, $tableName); $sql = "SELECT COUNT(*) as total FROM ({$innerSql}) sub FORMAT JSON"; - $result = $this->query($sql, $innerStatement->namedBindings ?? []); + $result = $this->queryRead($sql, $innerStatement->namedBindings ?? []); } else { $builder = $this->newBuilder($type) ->from($tableName) @@ -2560,7 +2697,7 @@ private function countFromTable(string $tenant, array $queries, string $type, ?i $statement = $builder->build(); $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - $result = $this->query($sql, $statement->namedBindings ?? []); + $result = $this->queryRead($sql, $statement->namedBindings ?? []); } return $this->decodeTotal($result); @@ -3223,7 +3360,7 @@ private function sumHybridDailyAndRaw(string $tenant, array $queries, array $pla FORMAT JSON "; - $result = $this->query($sql, array_merge($rawStatement->namedBindings ?? [], $dailyBindings)); + $result = $this->queryRead($sql, array_merge($rawStatement->namedBindings ?? [], $dailyBindings)); return $this->decodeTotal($result); } @@ -3254,7 +3391,7 @@ private function sumFromTable(string $tenant, array $queries, string $attribute, $statement = $builder->build(); $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - return $this->decodeTotal($this->query($sql, $statement->namedBindings ?? [])); + return $this->decodeTotal($this->queryRead($sql, $statement->namedBindings ?? [])); } /** @@ -3304,7 +3441,7 @@ public function findDaily(string $tenant, array $queries = []): array $statement = $builder->build(); $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - return $this->parseResults($this->query($sql, $statement->namedBindings ?? []), Usage::TYPE_EVENT); + return $this->parseResults($this->queryRead($sql, $statement->namedBindings ?? []), Usage::TYPE_EVENT); } /** @@ -3351,7 +3488,7 @@ private function sumDailyTotal(string $tenant, array $queries, string $attribute $statement = $builder->build(); $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - return $this->decodeTotal($this->query($sql, $statement->namedBindings ?? [])); + return $this->decodeTotal($this->queryRead($sql, $statement->namedBindings ?? [])); } /** @@ -3395,7 +3532,7 @@ public function sumDailyBatch(string $tenant, array $metrics, array $queries = [ $statement = $builder->build(); $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - $result = $this->query($sql, $statement->namedBindings ?? []); + $result = $this->queryRead($sql, $statement->namedBindings ?? []); $rows = $this->decodeRows($result); foreach ($rows as $row) { @@ -3545,7 +3682,7 @@ private function getTimeSeriesFromTable(string $tenant, array $metrics, string $ $statement = $builder->build(); $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - $result = $this->query($sql, $statement->namedBindings ?? []); + $result = $this->queryRead($sql, $statement->namedBindings ?? []); $rows = $this->decodeRows($result); // Initialize result structure @@ -3751,7 +3888,7 @@ private function getTotalFromGauges(string $tenant, string $metric, array $queri $statement = $builder->build(); $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - return $this->decodeTotal($this->query($sql, $statement->namedBindings ?? [])); + return $this->decodeTotal($this->queryRead($sql, $statement->namedBindings ?? [])); } /** @@ -3812,7 +3949,7 @@ public function getTotalBatch(string $tenant, array $metrics, array $queries = [ $statement = $builder->build(); $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - $result = $this->query($sql, $statement->namedBindings ?? []); + $result = $this->queryRead($sql, $statement->namedBindings ?? []); $rows = $this->decodeRows($result); foreach ($rows as $row) { diff --git a/tests/Usage/Adapter/ClickHouseReadCancellationTest.php b/tests/Usage/Adapter/ClickHouseReadCancellationTest.php new file mode 100644 index 0000000..3f7254f --- /dev/null +++ b/tests/Usage/Adapter/ClickHouseReadCancellationTest.php @@ -0,0 +1,144 @@ +newAdapter($client); + $adapter->setNextQueryId('pinned-id'); + + try { + $this->queryReadRaw($adapter, self::READ_SQL); + $this->fail('Expected the read to fail'); + } catch (Exception $e) { + $this->assertStringContainsString('Operation timed out', $e->getMessage()); + } + + $this->assertCount(2, $client->urls); + $this->assertStringContainsString('max_execution_time=25', $client->urls[0]); + $this->assertStringContainsString('query_id=pinned-id', $client->urls[0]); + $this->assertStringContainsString('KILL QUERY WHERE query_id = {queryId:String} ASYNC', $client->bodies[1]); + $this->assertStringContainsString('pinned-id', $client->bodies[1]); + } + + public function testFailedCancellationDoesNotMaskTheReadError(): void + { + $client = new ScriptedClient([ScriptedClient::TIMEOUT, ScriptedClient::TIMEOUT]); + $adapter = $this->newAdapter($client); + + try { + $this->queryReadRaw($adapter, self::READ_SQL); + $this->fail('Expected the read to fail'); + } catch (Exception $e) { + $this->assertStringContainsString(self::READ_SQL, $e->getMessage()); + } + + $this->assertCount(2, $client->urls); + } + + public function testUncappedReadIsNotCancelled(): void + { + $client = new ScriptedClient([ScriptedClient::TIMEOUT]); + $adapter = $this->newAdapter($client, readTimeout: null); + + try { + $this->queryReadRaw($adapter, self::READ_SQL); + $this->fail('Expected the read to fail'); + } catch (Exception $e) { + $this->assertStringContainsString('Operation timed out', $e->getMessage()); + } + + $this->assertCount(1, $client->urls); + $this->assertStringNotContainsString('max_execution_time', $client->urls[0]); + } + + public function testSettingsRejectionFallsBackToAnUncappedRead(): void + { + $readonly = ScriptedClient::response(403, "Code: 164. DB::Exception: Cannot modify 'max_execution_time' setting in readonly mode. (READONLY)"); + $client = new ScriptedClient([$readonly]); + $adapter = $this->newAdapter($client); + + $this->assertSame('{"data":[]}', $this->queryReadRaw($adapter, self::READ_SQL)); + $this->assertCount(2, $client->urls); + $this->assertStringContainsString('max_execution_time=25', $client->urls[0]); + $this->assertStringNotContainsString('max_execution_time', $client->urls[1]); + $this->assertStringNotContainsString('KILL QUERY', $client->bodies[1]); + + // The instance stops trying once refused. + $this->queryReadRaw($adapter, self::READ_SQL); + $this->assertCount(3, $client->urls); + $this->assertStringNotContainsString('max_execution_time', $client->urls[2]); + } + + public function testUnreachableServerIsNotCancelled(): void + { + $client = new ScriptedClient([ScriptedClient::UNREACHABLE]); + $adapter = $this->newAdapter($client); + + try { + $this->queryReadRaw($adapter, self::READ_SQL); + $this->fail('Expected the read to fail'); + } catch (Exception $e) { + $this->assertStringContainsString('Failed to connect', $e->getMessage()); + } + + // Nothing reached the server, so a KILL would only burn a second timeout. + $this->assertCount(1, $client->urls); + } + + public function testCancellationLeavesAnotherReadsPinnedIdAlone(): void + { + $client = new ScriptedClient(); + $adapter = $this->newAdapter($client); + $client->script = [ + // A concurrent caller pins its id while this read is failing. + function () use ($adapter): string { + $adapter->setNextQueryId('other-read-id'); + + return ScriptedClient::TIMEOUT; + }, + ]; + $adapter->setNextQueryId('failing-read-id'); + + try { + $this->queryReadRaw($adapter, self::READ_SQL); + $this->fail('Expected the read to fail'); + } catch (Exception $e) { + $this->assertStringContainsString('Operation timed out', $e->getMessage()); + } + + $this->assertStringContainsString('failing-read-id', $client->bodies[1]); + $this->assertStringNotContainsString('query_id=other-read-id', $client->urls[1]); + + $this->queryReadRaw($adapter, self::READ_SQL); + $this->assertStringContainsString('query_id=other-read-id', $client->urls[2]); + } +} diff --git a/tests/Usage/Adapter/ClickHouseReadTimeoutTest.php b/tests/Usage/Adapter/ClickHouseReadTimeoutTest.php new file mode 100644 index 0000000..d397a33 --- /dev/null +++ b/tests/Usage/Adapter/ClickHouseReadTimeoutTest.php @@ -0,0 +1,218 @@ +adapter = $this->newAdapter(); + $this->usage = new Usage($this->adapter); + $this->usage->setup(); + $this->usage->purge('1'); + + $this->usage->addBatch([ + ['tenant' => '1', 'metric' => 'read.timeout.metric', 'value' => 7], + ], Usage::TYPE_EVENT); + } + + protected function tearDown(): void + { + $this->usage->purge('1'); + } + + private function newAdapter(?int $readTimeout = 25, ?float $clientTimeout = null): ClickHouseAdapter + { + $client = $clientTimeout === null + ? null + : new Client((new CurlAdapter())->withTimeout($clientTimeout)); + + return new ClickHouseAdapter( + getenv('CLICKHOUSE_HOST') ?: 'clickhouse', + getenv('CLICKHOUSE_USER') ?: 'default', + getenv('CLICKHOUSE_PASSWORD') ?: 'clickhouse', + (int) (getenv('CLICKHOUSE_PORT') ?: 8123), + (bool) (getenv('CLICKHOUSE_SECURE') ?: false), + client: $client, + namespace: self::NAMESPACE, + database: getenv('CLICKHOUSE_DATABASE') ?: 'default', + sharedTables: true, + readTimeout: $readTimeout, + ); + } + + public function testReadCarriesExecutionCapAndGeneratedQueryId(): void + { + $this->find($this->adapter); + + $row = $this->lastQueryLogRow("query_kind = 'Select' AND query LIKE '%" . self::NAMESPACE . "_usage_events%'"); + + $this->assertSame('25', $row['max_execution_time'] ?? null); + $this->assertMatchesRegularExpression('/^[0-9a-f]{32}$/', (string) ($row['query_id'] ?? '')); + } + + public function testCallerSuppliedQueryIdSurvivesTheExecutionCap(): void + { + $queryId = 'read-timeout-' . bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $this->find($this->adapter); + + $row = $this->lastQueryLogRow("query_id = '{$queryId}'"); + + $this->assertSame('25', $row['max_execution_time'] ?? null); + } + + public function testNullReadTimeoutSendsNoExecutionCap(): void + { + $adapter = $this->newAdapter(readTimeout: null); + $queryId = 'read-timeout-off-' . bin2hex(random_bytes(8)); + $adapter->setNextQueryId($queryId); + $this->find($adapter); + + $row = $this->lastQueryLogRow("query_id = '{$queryId}'"); + + $this->assertSame('', $row['max_execution_time'] ?? null); + } + + public function testWritePathIsNotCapped(): void + { + $this->usage->addBatch([ + ['tenant' => '1', 'metric' => 'read.timeout.metric', 'value' => 11], + ], Usage::TYPE_EVENT); + + $row = $this->lastQueryLogRow("query_kind = 'Insert' AND query LIKE '%" . self::NAMESPACE . "_usage_events%'"); + + $this->assertSame('', $row['max_execution_time'] ?? null); + } + + public function testSlowReadIsAbortedByTheExecutionCap(): void + { + $adapter = $this->newAdapter(readTimeout: 1); + + $start = microtime(true); + try { + $this->queryReadRaw($adapter, self::SLOW_QUERY); + $this->fail('Expected the execution cap to abort the read'); + } catch (Exception $e) { + $this->assertStringContainsString('TIMEOUT_EXCEEDED', $e->getMessage()); + } + + // The query sleeps for 30s; anything close to that means ClickHouse ran + // it to completion instead of aborting. + $this->assertLessThan(10.0, microtime(true) - $start); + $this->assertSame(0, $this->runningSleepQueries()); + } + + public function testReadAbandonedByTheClientIsKilled(): void + { + // Cap above the client timeout so the socket dies first — the production + // shape, where the query would otherwise outlive the request. + $adapter = $this->newAdapter(readTimeout: 25, clientTimeout: 1.0); + $queryId = 'read-timeout-kill-' . bin2hex(random_bytes(8)); + $adapter->setNextQueryId($queryId); + + try { + $this->queryReadRaw($adapter, self::SLOW_QUERY); + $this->fail('Expected the client to give up on the read'); + } catch (Exception $e) { + $this->assertStringContainsString('ClickHouse query failed', $e->getMessage()); + } + + $this->assertSame(0, $this->runningQueries("query_id = '{$queryId}'")); + } + + public function testRejectsNonPositiveReadTimeout(): void + { + $this->expectException(Exception::class); + $this->expectExceptionMessage('Read timeout must be a positive number of seconds'); + + $this->newAdapter(readTimeout: 0); + } + + /** + * @return array<\Utopia\Usage\Metric> + */ + private function find(ClickHouseAdapter $adapter): array + { + return $adapter->find('1', [ + Query::equal('metric', ['read.timeout.metric']), + Query::limit(10), + ], Usage::TYPE_EVENT); + } + + /** + * Most recent finished query matching $where, with the settings we care about. + * + * @return array + */ + private function lastQueryLogRow(string $where): array + { + $this->queryRaw($this->adapter, 'SYSTEM FLUSH LOGS'); + + $sql = "SELECT query_id, Settings['max_execution_time'] AS max_execution_time " + . "FROM system.query_log WHERE type = 'QueryFinish' AND {$where} " + . 'ORDER BY event_time_microseconds DESC LIMIT 1 FORMAT JSON'; + + $json = json_decode($this->queryRaw($this->adapter, $sql), true); + $row = is_array($json) && is_array($json['data'] ?? null) ? ($json['data'][0] ?? null) : null; + $this->assertIsArray($row, "no query_log row for: {$where}"); + + $out = []; + foreach ($row as $key => $value) { + $out[(string) $key] = is_scalar($value) ? (string) $value : ''; + } + + return $out; + } + + private function runningSleepQueries(): int + { + return $this->runningQueries("query LIKE '%sleepEachRow%' AND query NOT LIKE '%system.processes%'"); + } + + /** + * Queries still running server-side. Cancellation is asynchronous, so poll + * briefly before declaring one abandoned. + */ + private function runningQueries(string $where): int + { + $sql = "SELECT count() AS running FROM system.processes WHERE {$where} FORMAT JSON"; + + $running = 0; + for ($attempt = 0; $attempt < 20; $attempt++) { + $json = json_decode($this->queryRaw($this->adapter, $sql), true); + $row = is_array($json) && is_array($json['data'] ?? null) ? ($json['data'][0] ?? null) : null; + $running = is_array($row) && is_numeric($row['running'] ?? null) ? (int) $row['running'] : 0; + if ($running === 0) { + return 0; + } + usleep(250_000); + } + + return $running; + } +} diff --git a/tests/Usage/Adapter/ClickHouseTestCase.php b/tests/Usage/Adapter/ClickHouseTestCase.php index 24d9ed4..0f59e32 100644 --- a/tests/Usage/Adapter/ClickHouseTestCase.php +++ b/tests/Usage/Adapter/ClickHouseTestCase.php @@ -57,4 +57,19 @@ protected function queryRaw(ClickHouseAdapter $adapter, string $sql, array $para $raw = $method->invoke($adapter, $sql, $params); return is_string($raw) ? $raw : ''; } + + /** + * Run raw SQL through the private queryRead() method — the read wrapper + * that applies the execution cap, the query_id and the cancellation. + * + * @param array $params + */ + protected function queryReadRaw(ClickHouseAdapter $adapter, string $sql, array $params = []): string + { + $reflection = new ReflectionClass($adapter); + $method = $reflection->getMethod('queryRead'); + $method->setAccessible(true); + $raw = $method->invoke($adapter, $sql, $params); + return is_string($raw) ? $raw : ''; + } } diff --git a/tests/Usage/Adapter/ScriptedClient.php b/tests/Usage/Adapter/ScriptedClient.php new file mode 100644 index 0000000..54367f1 --- /dev/null +++ b/tests/Usage/Adapter/ScriptedClient.php @@ -0,0 +1,67 @@ + URLs of the requests sent, in order */ + public array $urls = []; + + /** @var array Bodies of the requests sent, in order */ + public array $bodies = []; + + /** + * @param array $script One entry per + * request. TIMEOUT throws a socket timeout, UNREACHABLE a connection + * failure, and a Closure runs first so a test can mutate adapter state + * mid-flight before returning one of the above. + */ + public function __construct(public array $script = []) + { + } + + public static function response(int $status, string $body = ''): ResponseInterface + { + return new Response($status, '', new Stream($body)); + } + + public function sendRequest(RequestInterface $request): ResponseInterface + { + $this->urls[] = (string) $request->getUri(); + $this->bodies[] = (string) $request->getBody(); + + $next = array_shift($this->script); + + if ($next instanceof Closure) { + $next = $next(); + } + + if ($next === self::TIMEOUT) { + throw new TimeoutException($request, 'Operation timed out'); + } + + if ($next === self::UNREACHABLE) { + throw new ConnectionException($request, 'Failed to connect'); + } + + return $next instanceof ResponseInterface ? $next : self::response(200, '{"data":[]}'); + } +}