Skip to content

Commit 37223c9

Browse files
millotpFluf22
millotp
and
Fluf22
authored
feat(clients): cleanup after replaceAllObjects failure [skip-bc] (#3824)
Co-authored-by: Thomas Raffray <Fluf22@users.noreply.github.com>
1 parent 5553caf commit 37223c9

File tree

22 files changed

+595
-386
lines changed

22 files changed

+595
-386
lines changed

‎clients/algoliasearch-client-csharp/algoliasearch/Utils/SearchClientExtensions.cs

+33-24
Original file line numberDiff line numberDiff line change
@@ -494,37 +494,46 @@ public async Task<ReplaceAllObjectsResponse> ReplaceAllObjectsAsync<T>(string in
494494
var rnd = new Random();
495495
var tmpIndexName = $"{indexName}_tmp_{rnd.Next(100)}";
496496

497-
var copyResponse = await OperationIndexAsync(indexName,
498-
new OperationIndexParams(OperationType.Copy, tmpIndexName)
499-
{ Scope = [ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms] }, options, cancellationToken)
500-
.ConfigureAwait(false);
497+
try
498+
{
499+
var copyResponse = await OperationIndexAsync(indexName,
500+
new OperationIndexParams(OperationType.Copy, tmpIndexName)
501+
{ Scope = [ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms] }, options, cancellationToken)
502+
.ConfigureAwait(false);
501503

502-
var batchResponse = await ChunkedBatchAsync(tmpIndexName, objects, Action.AddObject, true, batchSize,
503-
options, cancellationToken).ConfigureAwait(false);
504+
var batchResponse = await ChunkedBatchAsync(tmpIndexName, objects, Action.AddObject, true, batchSize,
505+
options, cancellationToken).ConfigureAwait(false);
504506

505-
await WaitForTaskAsync(tmpIndexName, copyResponse.TaskID, requestOptions: options, ct: cancellationToken)
506-
.ConfigureAwait(false);
507+
await WaitForTaskAsync(tmpIndexName, copyResponse.TaskID, requestOptions: options, ct: cancellationToken)
508+
.ConfigureAwait(false);
507509

508-
copyResponse = await OperationIndexAsync(indexName,
509-
new OperationIndexParams(OperationType.Copy, tmpIndexName)
510-
{ Scope = [ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms] }, options, cancellationToken)
511-
.ConfigureAwait(false);
512-
await WaitForTaskAsync(tmpIndexName, copyResponse.TaskID, requestOptions: options, ct: cancellationToken)
513-
.ConfigureAwait(false);
510+
copyResponse = await OperationIndexAsync(indexName,
511+
new OperationIndexParams(OperationType.Copy, tmpIndexName)
512+
{ Scope = [ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms] }, options, cancellationToken)
513+
.ConfigureAwait(false);
514+
await WaitForTaskAsync(tmpIndexName, copyResponse.TaskID, requestOptions: options, ct: cancellationToken)
515+
.ConfigureAwait(false);
514516

515-
var moveResponse = await OperationIndexAsync(tmpIndexName,
516-
new OperationIndexParams(OperationType.Move, indexName), options, cancellationToken)
517-
.ConfigureAwait(false);
517+
var moveResponse = await OperationIndexAsync(tmpIndexName,
518+
new OperationIndexParams(OperationType.Move, indexName), options, cancellationToken)
519+
.ConfigureAwait(false);
518520

519-
await WaitForTaskAsync(tmpIndexName, moveResponse.TaskID, requestOptions: options, ct: cancellationToken)
520-
.ConfigureAwait(false);
521+
await WaitForTaskAsync(tmpIndexName, moveResponse.TaskID, requestOptions: options, ct: cancellationToken)
522+
.ConfigureAwait(false);
521523

522-
return new ReplaceAllObjectsResponse
524+
return new ReplaceAllObjectsResponse
525+
{
526+
CopyOperationResponse = copyResponse,
527+
MoveOperationResponse = moveResponse,
528+
BatchResponses = batchResponse
529+
};
530+
}
531+
catch (Exception ex)
523532
{
524-
CopyOperationResponse = copyResponse,
525-
MoveOperationResponse = moveResponse,
526-
BatchResponses = batchResponse
527-
};
533+
await DeleteIndexAsync(tmpIndexName, cancellationToken: cancellationToken).ConfigureAwait(false);
534+
535+
throw ex;
536+
}
528537
}
529538

530539
/// <inheritdoc/>

‎clients/algoliasearch-client-kotlin/client/src/commonMain/kotlin/com/algolia/client/extensions/SearchClient.kt

+48-35
Original file line numberDiff line numberDiff line change
@@ -472,46 +472,52 @@ public suspend fun SearchClient.replaceAllObjects(
472472
): ReplaceAllObjectsResponse {
473473
val tmpIndexName = "${indexName}_tmp_${Random.nextInt(from = 0, until = 100)}"
474474

475-
var copy = operationIndex(
476-
indexName = indexName,
477-
operationIndexParams = OperationIndexParams(
478-
operation = OperationType.Copy,
479-
destination = tmpIndexName,
480-
scope = listOf(ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms),
481-
),
482-
requestOptions = requestOptions,
483-
)
475+
try {
476+
var copy = operationIndex(
477+
indexName = indexName,
478+
operationIndexParams = OperationIndexParams(
479+
operation = OperationType.Copy,
480+
destination = tmpIndexName,
481+
scope = listOf(ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms),
482+
),
483+
requestOptions = requestOptions,
484+
)
484485

485-
val batchResponses = this.chunkedBatch(
486-
indexName = tmpIndexName,
487-
objects = objects,
488-
action = Action.AddObject,
489-
waitForTask = true,
490-
batchSize = batchSize,
491-
requestOptions = requestOptions,
492-
)
486+
val batchResponses = this.chunkedBatch(
487+
indexName = tmpIndexName,
488+
objects = objects,
489+
action = Action.AddObject,
490+
waitForTask = true,
491+
batchSize = batchSize,
492+
requestOptions = requestOptions,
493+
)
493494

494-
waitForTask(indexName = tmpIndexName, taskID = copy.taskID)
495+
waitForTask(indexName = tmpIndexName, taskID = copy.taskID)
495496

496-
copy = operationIndex(
497-
indexName = indexName,
498-
operationIndexParams = OperationIndexParams(
499-
operation = OperationType.Copy,
500-
destination = tmpIndexName,
501-
scope = listOf(ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms),
502-
),
503-
requestOptions = requestOptions,
504-
)
505-
waitForTask(indexName = tmpIndexName, taskID = copy.taskID)
497+
copy = operationIndex(
498+
indexName = indexName,
499+
operationIndexParams = OperationIndexParams(
500+
operation = OperationType.Copy,
501+
destination = tmpIndexName,
502+
scope = listOf(ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms),
503+
),
504+
requestOptions = requestOptions,
505+
)
506+
waitForTask(indexName = tmpIndexName, taskID = copy.taskID)
506507

507-
val move = operationIndex(
508-
indexName = tmpIndexName,
509-
operationIndexParams = OperationIndexParams(operation = OperationType.Move, destination = indexName),
510-
requestOptions = requestOptions,
511-
)
512-
waitForTask(indexName = tmpIndexName, taskID = move.taskID)
508+
val move = operationIndex(
509+
indexName = tmpIndexName,
510+
operationIndexParams = OperationIndexParams(operation = OperationType.Move, destination = indexName),
511+
requestOptions = requestOptions,
512+
)
513+
waitForTask(indexName = tmpIndexName, taskID = move.taskID)
514+
515+
return ReplaceAllObjectsResponse(copy, batchResponses, move)
516+
} catch (e: Exception) {
517+
deleteIndex(tmpIndexName)
513518

514-
return ReplaceAllObjectsResponse(copy, batchResponses, move)
519+
throw e
520+
}
515521
}
516522

517523
/**
@@ -542,6 +548,13 @@ public fun securedApiKeyRemainingValidity(apiKey: String): Duration {
542548
return validUntil - Clock.System.now()
543549
}
544550

551+
/**
552+
* Checks that an index exists.
553+
*
554+
* @param indexName The name of the index to check.
555+
* @return true if the index exists, false otherwise.
556+
* @throws AlgoliaApiException if an error occurs during the request.
557+
*/
545558
public suspend fun SearchClient.indexExists(indexName: String): Boolean {
546559
try {
547560
getSettings(indexName)

‎clients/algoliasearch-client-scala/src/main/scala/algoliasearch/extension/package.scala

+47-39
Original file line numberDiff line numberDiff line change
@@ -366,50 +366,58 @@ package object extension {
366366
)(implicit ec: ExecutionContext): Future[ReplaceAllObjectsResponse] = {
367367
val tmpIndexName = s"${indexName}_tmp_${scala.util.Random.nextInt(100)}"
368368

369-
for {
370-
copy <- client.operationIndex(
371-
indexName = indexName,
372-
operationIndexParams = OperationIndexParams(
373-
operation = OperationType.Copy,
374-
destination = tmpIndexName,
375-
scope = Some(Seq(ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms))
376-
),
377-
requestOptions = requestOptions
378-
)
369+
try {
370+
for {
371+
copy <- client.operationIndex(
372+
indexName = indexName,
373+
operationIndexParams = OperationIndexParams(
374+
operation = OperationType.Copy,
375+
destination = tmpIndexName,
376+
scope = Some(Seq(ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms))
377+
),
378+
requestOptions = requestOptions
379+
)
379380

380-
batchResponses <- chunkedBatch(
381-
indexName = tmpIndexName,
382-
objects = objects,
383-
action = Action.AddObject,
384-
waitForTasks = true,
385-
batchSize = batchSize,
386-
requestOptions = requestOptions
387-
)
381+
batchResponses <- chunkedBatch(
382+
indexName = tmpIndexName,
383+
objects = objects,
384+
action = Action.AddObject,
385+
waitForTasks = true,
386+
batchSize = batchSize,
387+
requestOptions = requestOptions
388+
)
388389

389-
_ <- client.waitTask(indexName = tmpIndexName, taskID = copy.taskID, requestOptions = requestOptions)
390+
_ <- client.waitTask(indexName = tmpIndexName, taskID = copy.taskID, requestOptions = requestOptions)
390391

391-
copy <- client.operationIndex(
392-
indexName = indexName,
393-
operationIndexParams = OperationIndexParams(
394-
operation = OperationType.Copy,
395-
destination = tmpIndexName,
396-
scope = Some(Seq(ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms))
397-
),
398-
requestOptions = requestOptions
399-
)
400-
_ <- client.waitTask(indexName = tmpIndexName, taskID = copy.taskID, requestOptions = requestOptions)
392+
copy <- client.operationIndex(
393+
indexName = indexName,
394+
operationIndexParams = OperationIndexParams(
395+
operation = OperationType.Copy,
396+
destination = tmpIndexName,
397+
scope = Some(Seq(ScopeType.Settings, ScopeType.Rules, ScopeType.Synonyms))
398+
),
399+
requestOptions = requestOptions
400+
)
401+
_ <- client.waitTask(indexName = tmpIndexName, taskID = copy.taskID, requestOptions = requestOptions)
401402

402-
move <- client.operationIndex(
403-
indexName = tmpIndexName,
404-
operationIndexParams = OperationIndexParams(operation = OperationType.Move, destination = indexName),
405-
requestOptions = requestOptions
403+
move <- client.operationIndex(
404+
indexName = tmpIndexName,
405+
operationIndexParams = OperationIndexParams(operation = OperationType.Move, destination = indexName),
406+
requestOptions = requestOptions
407+
)
408+
_ <- client.waitTask(indexName = tmpIndexName, taskID = move.taskID, requestOptions = requestOptions)
409+
} yield ReplaceAllObjectsResponse(
410+
copyOperationResponse = copy,
411+
batchResponses = batchResponses,
412+
moveOperationResponse = move
406413
)
407-
_ <- client.waitTask(indexName = tmpIndexName, taskID = move.taskID, requestOptions = requestOptions)
408-
} yield ReplaceAllObjectsResponse(
409-
copyOperationResponse = copy,
410-
batchResponses = batchResponses,
411-
moveOperationResponse = move
412-
)
414+
} catch {
415+
case e : Throwable => {
416+
client.deleteIndex(tmpIndexName)
417+
418+
throw e
419+
}
420+
}
413421
}
414422

415423
/** Check if an index exists.

‎clients/algoliasearch-client-swift/Sources/Search/Extra/SearchClientExtension.swift

+48-42
Original file line numberDiff line numberDiff line change
@@ -559,51 +559,57 @@ public extension SearchClient {
559559
) async throws -> ReplaceAllObjectsResponse {
560560
let tmpIndexName = "\(indexName)_tmp_\(Int.random(in: 1_000_000 ..< 10_000_000))"
561561

562-
var copyOperationResponse = try await operationIndex(
563-
indexName: indexName,
564-
operationIndexParams: OperationIndexParams(
565-
operation: .copy,
566-
destination: tmpIndexName,
567-
scope: [.settings, .rules, .synonyms]
568-
),
569-
requestOptions: requestOptions
570-
)
562+
do {
563+
var copyOperationResponse = try await operationIndex(
564+
indexName: indexName,
565+
operationIndexParams: OperationIndexParams(
566+
operation: .copy,
567+
destination: tmpIndexName,
568+
scope: [.settings, .rules, .synonyms]
569+
),
570+
requestOptions: requestOptions
571+
)
571572

572-
let batchResponses = try await self.chunkedBatch(
573-
indexName: tmpIndexName,
574-
objects: objects,
575-
waitForTasks: true,
576-
batchSize: batchSize,
577-
requestOptions: requestOptions
578-
)
579-
try await self.waitForTask(indexName: tmpIndexName, taskID: copyOperationResponse.taskID)
573+
let batchResponses = try await self.chunkedBatch(
574+
indexName: tmpIndexName,
575+
objects: objects,
576+
waitForTasks: true,
577+
batchSize: batchSize,
578+
requestOptions: requestOptions
579+
)
580+
try await self.waitForTask(indexName: tmpIndexName, taskID: copyOperationResponse.taskID)
580581

581-
copyOperationResponse = try await operationIndex(
582-
indexName: indexName,
583-
operationIndexParams: OperationIndexParams(
584-
operation: .copy,
585-
destination: tmpIndexName,
586-
scope: [.settings, .rules, .synonyms]
587-
),
588-
requestOptions: requestOptions
589-
)
590-
try await self.waitForTask(indexName: tmpIndexName, taskID: copyOperationResponse.taskID)
591-
592-
let moveOperationResponse = try await self.operationIndex(
593-
indexName: tmpIndexName,
594-
operationIndexParams: OperationIndexParams(
595-
operation: .move,
596-
destination: indexName
597-
),
598-
requestOptions: requestOptions
599-
)
600-
try await self.waitForTask(indexName: tmpIndexName, taskID: moveOperationResponse.taskID)
582+
copyOperationResponse = try await operationIndex(
583+
indexName: indexName,
584+
operationIndexParams: OperationIndexParams(
585+
operation: .copy,
586+
destination: tmpIndexName,
587+
scope: [.settings, .rules, .synonyms]
588+
),
589+
requestOptions: requestOptions
590+
)
591+
try await self.waitForTask(indexName: tmpIndexName, taskID: copyOperationResponse.taskID)
601592

602-
return ReplaceAllObjectsResponse(
603-
copyOperationResponse: copyOperationResponse,
604-
batchResponses: batchResponses,
605-
moveOperationResponse: moveOperationResponse
606-
)
593+
let moveOperationResponse = try await self.operationIndex(
594+
indexName: tmpIndexName,
595+
operationIndexParams: OperationIndexParams(
596+
operation: .move,
597+
destination: indexName
598+
),
599+
requestOptions: requestOptions
600+
)
601+
try await self.waitForTask(indexName: tmpIndexName, taskID: moveOperationResponse.taskID)
602+
603+
return ReplaceAllObjectsResponse(
604+
copyOperationResponse: copyOperationResponse,
605+
batchResponses: batchResponses,
606+
moveOperationResponse: moveOperationResponse
607+
)
608+
} catch {
609+
_ = try? await self.deleteIndex(indexName: tmpIndexName)
610+
611+
throw error
612+
}
607613
}
608614

609615
/// Generate a secured API key

‎scripts/cts/runCts.ts

+2
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import { printBenchmarkReport } from './testServer/benchmark.js';
99
import { assertChunkWrapperValid } from './testServer/chunkWrapper.js';
1010
import { startTestServer } from './testServer/index.js';
1111
import { assertValidReplaceAllObjects } from './testServer/replaceAllObjects.js';
12+
import { assertValidReplaceAllObjectsFailed } from './testServer/replaceAllObjectsFailed.js';
1213
import { assertValidTimeouts } from './testServer/timeout.js';
1314
import { assertValidWaitForApiKey } from './testServer/waitFor.js';
1415

@@ -152,6 +153,7 @@ export async function runCts(
152153
assertValidTimeouts(languages.length);
153154
assertChunkWrapperValid(languages.length - skip('dart') - skip('scala'));
154155
assertValidReplaceAllObjects(languages.length - skip('dart') - skip('scala'));
156+
assertValidReplaceAllObjectsFailed(languages.length - skip('dart') - skip('scala'));
155157
assertValidWaitForApiKey(languages.length - skip('dart') - skip('scala'));
156158
}
157159
if (withBenchmarkServer) {

0 commit comments

Comments
 (0)