Skip to content

fix(clickhouse): cancel reads the client has abandoned #33

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
lohanidamodar wants to merge 3 commits into from
Closed

fix(clickhouse): cancel reads the client has abandoned #33

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e653563
fix(clickhouse): cancel reads the client has abandoned
lohanidamodar Aug 30, 2026
b5f08f8
fix(clickhouse): own the read query_id instead of staging it
lohanidamodar Aug 30, 2026
e465ca5
docs(readme): document the ClickHouse read timeout
lohanidamodar Aug 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
22 changes: 22 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
177 changes: 157 additions & 20 deletions src/Usage/Adapter/ClickHouse.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<string, string> Maps interval strings to ClickHouse time functions */
private const INTERVAL_FUNCTIONS = [
'1h' => 'toStartOfHour',
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand All @@ -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');
}
Expand All @@ -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
Expand Down Expand Up @@ -514,13 +546,14 @@ private function buildErrorMessage(string $baseMessage, ?string $table = null, ?
*
* @param string $sql
* @param array<string, mixed> $params
* @param array<string, string> $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';

Expand All @@ -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++;
Expand All @@ -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<string, mixed> $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);
Comment thread
lohanidamodar marked this conversation as resolved.
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.
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -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)
Expand All @@ -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);
Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -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 ?? []));
}

/**
Expand Down Expand Up @@ -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);
}

/**
Expand Down Expand Up @@ -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 ?? []));
}

/**
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 ?? []));
}

/**
Expand Down Expand Up @@ -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) {
Expand Down
Loading
Loading