Skip to content

Commit 929ab40

Browse files
lohanidamodarlohanidamodar
committed
feat: implement automatic retry logic with exponential backoff for ClickHouse operations
1 parent b5605bd commit 929ab40

1 file changed

Lines changed: 167 additions & 143 deletions

File tree

‎src/Usage/Adapter/ClickHouse.php‎

Lines changed: 167 additions & 143 deletions
Original file line numberDiff line numberDiff line change
@@ -618,6 +618,47 @@ private function setOperationContext(?string $context): void
618618
$this->operationContext = $context;
619619
}
620620

621+
/**
622+
* Execute an operation with automatic retry logic and exponential backoff.
623+
*
624+
* @template T
625+
* @param callable(int): T $operation Callback that performs the operation, receives attempt number
626+
* @param callable(Exception, int|null): bool $shouldRetry Callback to determine if error is retryable
627+
* @param callable(Exception, int): Exception $buildException Callback to build final exception with context
628+
* @return T The result from the operation
629+
* @throws Exception
630+
*/
631+
private function executeWithRetry(callable $operation, callable $shouldRetry, callable $buildException): mixed
632+
{
633+
$attempt = 0;
634+
$lastException = null;
635+
636+
while ($attempt <= $this->maxRetries) {
637+
try {
638+
return $operation($attempt);
639+
} catch (Exception $e) {
640+
$lastException = $e;
641+
642+
// Check if we should retry
643+
if ($attempt < $this->maxRetries && $shouldRetry($e, $attempt)) {
644+
$attempt++;
645+
$delay = $this->retryDelay * (2 ** ($attempt - 1)); // Exponential backoff
646+
usleep($delay * 1000); // Convert ms to microseconds
647+
continue;
648+
}
649+
650+
// Not retryable or max retries reached
651+
throw $buildException($e, $attempt);
652+
}
653+
}
654+
655+
// Should never reach here, but just in case
656+
throw $buildException(
657+
$lastException ?? new Exception('Unknown error occurred'),
658+
$this->maxRetries
659+
);
660+
}
661+
621662
/**
622663
* Build a contextual error message with operation, table, and query info.
623664
*
@@ -671,42 +712,40 @@ private function buildErrorMessage(string $baseMessage, ?string $table = null, ?
671712
*/
672713
private function query(string $sql, array $params = []): string
673714
{
674-
$attempt = 0;
675-
$lastException = null;
676-
677-
while ($attempt <= $this->maxRetries) {
678-
$startTime = microtime(true);
679-
$scheme = $this->secure ? 'https' : 'http';
680-
$url = "{$scheme}://{$this->host}:{$this->port}/";
681-
682-
// Update the database header for each query (in case setDatabase was called)
683-
$this->client->addHeader('X-ClickHouse-Database', $this->database);
684-
685-
// Enable keep-alive for connection pooling
686-
if ($this->enableKeepAlive) {
687-
$this->client->addHeader('Connection', 'keep-alive');
688-
} else {
689-
$this->client->addHeader('Connection', 'close');
690-
}
715+
return $this->executeWithRetry(
716+
// Operation to execute
717+
function (int $attempt) use ($sql, $params): string {
718+
$startTime = microtime(true);
719+
$scheme = $this->secure ? 'https' : 'http';
720+
$url = "{$scheme}://{$this->host}:{$this->port}/";
721+
722+
// Update the database header for each query (in case setDatabase was called)
723+
$this->client->addHeader('X-ClickHouse-Database', $this->database);
724+
725+
// Enable keep-alive for connection pooling
726+
if ($this->enableKeepAlive) {
727+
$this->client->addHeader('Connection', 'keep-alive');
728+
} else {
729+
$this->client->addHeader('Connection', 'close');
730+
}
691731

692-
// Enable compression if configured
693-
if ($this->enableCompression) {
694-
$this->client->addHeader('Accept-Encoding', 'gzip');
695-
}
732+
// Enable compression if configured
733+
if ($this->enableCompression) {
734+
$this->client->addHeader('Accept-Encoding', 'gzip');
735+
}
696736

697-
// Track request count for statistics (only on first attempt)
698-
if ($attempt === 0) {
699-
$this->requestCount++;
700-
}
737+
// Track request count for statistics (only on first attempt)
738+
if ($attempt === 0) {
739+
$this->requestCount++;
740+
}
701741

702-
// Build multipart form data body with query and parameters
703-
// The Fetch client will automatically encode arrays as multipart/form-data
704-
$body = ['query' => $sql];
705-
foreach ($params as $key => $value) {
706-
$body['param_' . $key] = $this->formatParamValue($value);
707-
}
742+
// Build multipart form data body with query and parameters
743+
// The Fetch client will automatically encode arrays as multipart/form-data
744+
$body = ['query' => $sql];
745+
foreach ($params as $key => $value) {
746+
$body['param_' . $key] = $this->formatParamValue($value);
747+
}
708748

709-
try {
710749
$response = $this->client->fetch(
711750
url: $url,
712751
method: Client::METHOD_POST,
@@ -722,46 +761,42 @@ private function query(string $sql, array $params = []): string
722761
$errorMsg = $this->buildErrorMessage($baseError, null, $sql);
723762
$this->logQuery($sql, $params, $duration, false, $errorMsg, $attempt);
724763

725-
// Check if error is retryable
726-
if ($attempt < $this->maxRetries && $this->isRetryableError($httpCode, $baseError)) {
727-
$attempt++;
728-
$delay = $this->retryDelay * (2 ** ($attempt - 1)); // Exponential backoff
729-
usleep($delay * 1000); // Convert ms to microseconds
730-
continue;
731-
}
732-
733-
throw new Exception($errorMsg);
764+
throw new Exception($errorMsg . '|HTTP_CODE:' . $httpCode);
734765
}
735766

736767
$body = $response->getBody();
737768
$result = is_string($body) ? $body : '';
738769
$duration = microtime(true) - $startTime;
739770
$this->logQuery($sql, $params, $duration, true, null, $attempt);
740771
return $result;
741-
} catch (Exception $e) {
742-
$duration = microtime(true) - $startTime;
743-
$this->logQuery($sql, $params, $duration, false, $e->getMessage(), $attempt);
744-
$lastException = $e;
772+
},
773+
// Should retry predicate
774+
function (Exception $e, ?int $httpCode): bool {
775+
// Extract HTTP code from exception message if embedded
776+
$exceptionHttpCode = null;
777+
if (preg_match('/\|HTTP_CODE:(\d+)$/', $e->getMessage(), $matches)) {
778+
$exceptionHttpCode = (int) $matches[1];
779+
}
745780

746-
// Check if error is retryable
747-
if ($attempt < $this->maxRetries && $this->isRetryableError(null, $e->getMessage())) {
748-
$attempt++;
749-
$delay = $this->retryDelay * (2 ** ($attempt - 1)); // Exponential backoff
750-
usleep($delay * 1000); // Convert ms to microseconds
751-
continue;
781+
return $this->isRetryableError($exceptionHttpCode, $e->getMessage());
782+
},
783+
// Build final exception
784+
function (Exception $e, int $attempt) use ($sql): Exception {
785+
// Clean up HTTP code marker if present
786+
$cleanMessage = preg_replace('/\|HTTP_CODE:\d+$/', '', $e->getMessage());
787+
$cleanMessage = is_string($cleanMessage) ? $cleanMessage : $e->getMessage();
788+
789+
// If message already has context, return as-is
790+
if (strpos($cleanMessage, '[Operation:') !== false) {
791+
return new Exception($cleanMessage, 0, $e);
752792
}
753793

754-
// Preserve the original exception context for better debugging
755-
$baseError = "ClickHouse query execution failed after " . ($attempt + 1) . " attempt(s): {$e->getMessage()}";
794+
// Otherwise, build context
795+
$baseError = "ClickHouse query execution failed after " . ($attempt + 1) . " attempt(s): {$cleanMessage}";
756796
$errorMsg = $this->buildErrorMessage($baseError, null, $sql);
757-
throw new Exception($errorMsg, 0, $e);
797+
return new Exception($errorMsg, 0, $e);
758798
}
759-
}
760-
761-
// Should never reach here, but just in case
762-
$baseError = "ClickHouse query execution failed after " . ($this->maxRetries + 1) . " attempt(s)";
763-
$errorMsg = $this->buildErrorMessage($baseError, null, $sql);
764-
throw new Exception($errorMsg, 0, $lastException);
799+
);
765800
}
766801

767802
/**
@@ -779,108 +814,97 @@ private function insert(string $table, array $data): void
779814
return;
780815
}
781816

782-
$attempt = 0;
783-
$lastException = null;
817+
$this->executeWithRetry(
818+
// Operation to execute
819+
function (int $attempt) use ($table, $data): void {
820+
$startTime = microtime(true);
821+
$scheme = $this->secure ? 'https' : 'http';
822+
$escapedTable = $this->escapeIdentifier($table);
823+
$url = "{$scheme}://{$this->host}:{$this->port}/?query=INSERT+INTO+{$escapedTable}+FORMAT+JSONEachRow";
784824

785-
while ($attempt <= $this->maxRetries) {
786-
$startTime = microtime(true);
787-
$scheme = $this->secure ? 'https' : 'http';
788-
$escapedTable = $this->escapeIdentifier($table);
789-
$url = "{$scheme}://{$this->host}:{$this->port}/?query=INSERT+INTO+{$escapedTable}+FORMAT+JSONEachRow";
790-
791-
// Update the database header
792-
$this->client->addHeader('X-ClickHouse-Database', $this->database);
793-
$this->client->addHeader('Content-Type', 'application/x-ndjson');
794-
795-
// Enable keep-alive for connection pooling
796-
if ($this->enableKeepAlive) {
797-
$this->client->addHeader('Connection', 'keep-alive');
798-
} else {
799-
$this->client->addHeader('Connection', 'close');
800-
}
825+
// Update the database header
826+
$this->client->addHeader('X-ClickHouse-Database', $this->database);
827+
$this->client->addHeader('Content-Type', 'application/x-ndjson');
801828

802-
// Enable compression if configured
803-
if ($this->enableCompression) {
804-
$this->client->addHeader('Accept-Encoding', 'gzip');
805-
}
829+
// Enable keep-alive for connection pooling
830+
if ($this->enableKeepAlive) {
831+
$this->client->addHeader('Connection', 'keep-alive');
832+
} else {
833+
$this->client->addHeader('Connection', 'close');
834+
}
806835

807-
// Track request count for statistics (only on first attempt)
808-
if ($attempt === 0) {
809-
$this->requestCount++;
810-
}
836+
// Enable compression if configured
837+
if ($this->enableCompression) {
838+
$this->client->addHeader('Accept-Encoding', 'gzip');
839+
}
811840

812-
// Join JSON strings with newlines
813-
$body = implode("\n", $data);
841+
// Track request count for statistics (only on first attempt)
842+
if ($attempt === 0) {
843+
$this->requestCount++;
844+
}
814845

815-
$sql = "INSERT INTO {$escapedTable} FORMAT JSONEachRow";
816-
$params = ['rows' => count($data), 'bytes' => strlen($body)];
846+
// Join JSON strings with newlines
847+
$body = implode("\n", $data);
817848

818-
try {
819-
$response = $this->client->fetch(
820-
url: $url,
821-
method: Client::METHOD_POST,
822-
body: $body
823-
);
849+
$sql = "INSERT INTO {$escapedTable} FORMAT JSONEachRow";
850+
$params = ['rows' => count($data), 'bytes' => strlen($body)];
824851

825-
$httpCode = $response->getStatusCode();
852+
try {
853+
$response = $this->client->fetch(
854+
url: $url,
855+
method: Client::METHOD_POST,
856+
body: $body
857+
);
826858

827-
if ($httpCode !== 200) {
828-
$bodyStr = $response->getBody();
829-
$bodyStr = is_string($bodyStr) ? $bodyStr : '';
830-
$duration = microtime(true) - $startTime;
831-
$rowCount = count($data);
832-
$baseError = "ClickHouse insert failed with HTTP {$httpCode}: {$bodyStr}";
833-
$errorMsg = $this->buildErrorMessage($baseError, $table, "INSERT INTO {$table} ({$rowCount} rows)");
834-
$this->logQuery($sql, $params, $duration, false, $errorMsg, $attempt);
859+
$httpCode = $response->getStatusCode();
835860

836-
// Clean up Content-Type before retry
837-
$this->client->removeHeader('Content-Type');
861+
if ($httpCode !== 200) {
862+
$bodyStr = $response->getBody();
863+
$bodyStr = is_string($bodyStr) ? $bodyStr : '';
864+
$duration = microtime(true) - $startTime;
865+
$rowCount = count($data);
866+
$baseError = "ClickHouse insert failed with HTTP {$httpCode}: {$bodyStr}";
867+
$errorMsg = $this->buildErrorMessage($baseError, $table, "INSERT INTO {$table} ({$rowCount} rows)");
868+
$this->logQuery($sql, $params, $duration, false, $errorMsg, $attempt);
838869

839-
// Check if error is retryable
840-
if ($attempt < $this->maxRetries && $this->isRetryableError($httpCode, $baseError)) {
841-
$attempt++;
842-
$delay = $this->retryDelay * (2 ** ($attempt - 1)); // Exponential backoff
843-
usleep($delay * 1000); // Convert ms to microseconds
844-
continue;
870+
throw new Exception($errorMsg . '|HTTP_CODE:' . $httpCode);
845871
}
846872

847-
throw new Exception($errorMsg);
873+
$duration = microtime(true) - $startTime;
874+
$this->logQuery($sql, $params, $duration, true, null, $attempt);
875+
} finally {
876+
// Always clean up Content-Type header
877+
$this->client->removeHeader('Content-Type');
878+
}
879+
},
880+
// Should retry predicate
881+
function (Exception $e, ?int $httpCode): bool {
882+
// Extract HTTP code from exception message if embedded
883+
$exceptionHttpCode = null;
884+
if (preg_match('/\|HTTP_CODE:(\d+)$/', $e->getMessage(), $matches)) {
885+
$exceptionHttpCode = (int) $matches[1];
848886
}
849887

850-
$duration = microtime(true) - $startTime;
851-
$this->logQuery($sql, $params, $duration, true, null, $attempt);
852-
853-
// Clean up Content-Type after successful insert
854-
$this->client->removeHeader('Content-Type');
855-
return;
856-
} catch (Exception $e) {
857-
$duration = microtime(true) - $startTime;
858-
$this->logQuery($sql, $params, $duration, false, $e->getMessage(), $attempt);
859-
$lastException = $e;
860-
861-
// Clean up Content-Type before retry
862-
$this->client->removeHeader('Content-Type');
863-
864-
// Check if error is retryable
865-
if ($attempt < $this->maxRetries && $this->isRetryableError(null, $e->getMessage())) {
866-
$attempt++;
867-
$delay = $this->retryDelay * (2 ** ($attempt - 1)); // Exponential backoff
868-
usleep($delay * 1000); // Convert ms to microseconds
869-
continue;
888+
return $this->isRetryableError($exceptionHttpCode, $e->getMessage());
889+
},
890+
// Build final exception
891+
function (Exception $e, int $attempt) use ($table, $data): Exception {
892+
// Clean up HTTP code marker if present
893+
$cleanMessage = preg_replace('/\|HTTP_CODE:\d+$/', '', $e->getMessage());
894+
$cleanMessage = is_string($cleanMessage) ? $cleanMessage : $e->getMessage();
895+
896+
// If message already has context, return as-is
897+
if (strpos($cleanMessage, '[Operation:') !== false) {
898+
return new Exception($cleanMessage, 0, $e);
870899
}
871900

901+
// Otherwise, build context
872902
$rowCount = count($data);
873-
$baseError = "ClickHouse insert execution failed after " . ($attempt + 1) . " attempt(s): {$e->getMessage()}";
903+
$baseError = "ClickHouse insert execution failed after " . ($attempt + 1) . " attempt(s): {$cleanMessage}";
874904
$errorMsg = $this->buildErrorMessage($baseError, $table, "INSERT INTO {$table} ({$rowCount} rows)");
875-
throw new Exception($errorMsg, 0, $e);
905+
return new Exception($errorMsg, 0, $e);
876906
}
877-
}
878-
879-
// Should never reach here, but just in case
880-
$rowCount = count($data);
881-
$baseError = "ClickHouse insert execution failed after " . ($this->maxRetries + 1) . " attempt(s)";
882-
$errorMsg = $this->buildErrorMessage($baseError, $table, "INSERT INTO {$table} ({$rowCount} rows)");
883-
throw new Exception($errorMsg, 0, $lastException);
907+
);
884908
}
885909

886910
/**

0 commit comments

Comments
 (0)