fix(go): retry strategy (#5159)

Co-authored-by: Antoine GILLES <[email protected]>

Author: shortcuts

clients/algoliasearch-client-go/algolia/transport/transport.go

@@ -44,6 +44,27 @@ func New(cfg Configuration) *Transport {
 	return transport
 }
 
+func prepareRetryableRequest(req *http.Request) (*http.Request, error) {
+	// Read the original body
+	if req.Body == nil {
+		return req, nil // Nothing to do if there's no body
+	}
+
+	bodyBytes, err := io.ReadAll(req.Body)
+	if err != nil {
+		return nil, fmt.Errorf("cannot read body: %v", err)
+	}
+	_ = req.Body.Close() // close the original body
+
+	// Set up GetBody to recreate the body for retries
+	req.Body = io.NopCloser(bytes.NewReader(bodyBytes))
+	req.GetBody = func() (io.ReadCloser, error) {
+		return io.NopCloser(bytes.NewReader(bodyBytes)), nil
+	}
+
+	return req, nil
+}
+
 func (t *Transport) Request(ctx context.Context, req *http.Request, k call.Kind, c RequestConfiguration) (*http.Response, []byte, error) {
 	var intermediateNetworkErrors []error
 
@@ -52,7 +73,13 @@ func (t *Transport) Request(ctx context.Context, req *http.Request, k call.Kind,
 		req.Header.Add("Content-Encoding", "gzip")
 	}
 
-	for _, h := range t.retryStrategy.GetTryableHosts(k) {
+	// Prepare the request to be retryable.
+	req, err := prepareRetryableRequest(req)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	for i, h := range t.retryStrategy.GetTryableHosts(k) {
 		// Handle per-request timeout by using a context with timeout.
 		// Note that because we are in a loop, the cancel() callback cannot be
 		// deferred. Instead, we call it precisely after the end of each loop or
@@ -62,8 +89,17 @@ func (t *Transport) Request(ctx context.Context, req *http.Request, k call.Kind,
 		var (
 			ctxTimeout     time.Duration
 			connectTimeout time.Duration
+			err            error
 		)
 
+		// Reassign a fresh body for the retry
+		if i > 0 && req.GetBody != nil {
+			req.Body, err = req.GetBody()
+			if err != nil {
+				break
+			}
+		}
+
 		switch {
 		case k == call.Read && c.ReadTimeout != nil:
 			ctxTimeout = *c.ReadTimeout
@@ -113,6 +149,9 @@ func (t *Transport) Request(ctx context.Context, req *http.Request, k call.Kind,
 		default:
 			if err != nil {
 				intermediateNetworkErrors = append(intermediateNetworkErrors, err)
+			} else if res != nil {
+				msg := fmt.Sprintf("cannot perform request:\n\tStatusCode=%d\n\tmethod=%s\n\turl=%s\n\t", res.StatusCode, req.Method, req.URL)
+				intermediateNetworkErrors = append(intermediateNetworkErrors, errors.New(msg))
 			}
 			if res != nil && res.Body != nil {
 				if err = res.Body.Close(); err != nil {

scripts/cts/runCts.ts

@@ -8,6 +8,7 @@ import type { Language } from '../types.ts';
 import { assertValidAccountCopyIndex } from './testServer/accountCopyIndex.ts';
 import { printBenchmarkReport } from './testServer/benchmark.ts';
 import { assertChunkWrapperValid } from './testServer/chunkWrapper.ts';
+import { assertValidErrors } from './testServer/error.ts';
 import { startTestServer } from './testServer/index.ts';
 import { assertPushMockValid } from './testServer/pushMock.ts';
 import { assertValidReplaceAllObjects } from './testServer/replaceAllObjects.ts';
@@ -152,6 +153,7 @@ export async function runCts(
     const skip = (lang: Language): number => (languages.includes(lang) ? 1 : 0);
     const only = skip;
 
+    assertValidErrors(languages.length);
     assertValidTimeouts(languages.length);
     assertChunkWrapperValid(languages.length - skip('dart'));
     assertValidReplaceAllObjects(languages.length - skip('dart'));

scripts/cts/testServer/error.ts

@@ -0,0 +1,60 @@
+import type { Server } from 'http';
+
+import { expect } from 'chai';
+import type express from 'express';
+
+import { setupServer } from './index.ts';
+
+const errorState: Record<string, { errorCount: number; maxError: number }> = {};
+
+export function assertValidErrors(expectedCount: number): void {
+  // assert that the retry strategy uses the correct timings, by checking the time between each request, and how long each request took before being timed out
+  // there should be no delay between requests, only an increase in error.
+  if (Object.keys(errorState).length !== expectedCount) {
+    throw new Error(`Expected ${expectedCount} error(s)`);
+  }
+
+  for (const [lang, state] of Object.entries(errorState)) {
+    // python has sync and async tests
+    if (lang === 'python') {
+      expect(state.errorCount).to.equal(state.maxError * 2);
+
+      return;
+    }
+
+    expect(state.errorCount).to.equal(state.maxError);
+  }
+}
+
+function addRoutes(app: express.Express): void {
+  app.post('/1/test/error/:lang', (req, res) => {
+    const lang = req.params.lang;
+    if (!errorState[lang]) {
+      errorState[lang] = {
+        errorCount: 0,
+        maxError: 3,
+      };
+    }
+
+    errorState[lang].errorCount++;
+
+    if (errorState[lang].errorCount % errorState[lang].maxError !== 0) {
+      res.status(500).json({ message: 'error test server response' });
+      return;
+    }
+
+    res.status(200).json({ status: 'ok' });
+  });
+}
+
+export function errorServer(): Promise<Server> {
+  return setupServer('error', 6671, addRoutes);
+}
+
+export function errorServerRetriedOnce(): Promise<Server> {
+  return setupServer('errorRetriedOnce', 6672, addRoutes);
+}
+
+export function errorServerRetriedTwice(): Promise<Server> {
+  return setupServer('errorRetriedTwice', 6673, addRoutes);
+}

scripts/cts/testServer/index.ts

@@ -12,6 +12,7 @@ import { algoliaMockServer } from './algoliaMock.ts';
 import { apiKeyServer } from './apiKey.ts';
 import { benchmarkServer } from './benchmark.ts';
 import { chunkWrapperServer } from './chunkWrapper.ts';
+import { errorServer, errorServerRetriedOnce, errorServerRetriedTwice } from './error.ts';
 import { gzipServer } from './gzip.ts';
 import { pushMockServer } from './pushMock.ts';
 import { replaceAllObjectsServer } from './replaceAllObjects.ts';
@@ -27,6 +28,9 @@ export async function startTestServer(suites: Record<CTSType, boolean>): Promise
   if (suites.client) {
     toStart.push(
       timeoutServer(),
+      errorServer(),
+      errorServerRetriedOnce(),
+      errorServerRetriedTwice(),
       gzipServer(),
       timeoutServerBis(),
       accountCopyIndexServer(),

scripts/cts/testServer/replaceAllObjectsWithTransformation.ts

@@ -88,7 +88,6 @@ function addRoutes(app: Express): void {
       case 'move': {
         const lang = req.body.destination.replace('cts_e2e_replace_all_objects_with_transformation_', '');
         expect(raowtState).to.include.keys(lang);
-        console.log(raowtState[lang]);
         expect(raowtState[lang]).to.deep.equal({
           copyCount: 2,
           pushCount: 4,

tests/CTS/client/search/api.json

@@ -109,7 +109,7 @@
     ]
   },
   {
-    "testName": "tests the retry strategy error",
+    "testName": "tests the retry strategy on timeout",
     "autoCreateClient": false,
     "steps": [
       {
@@ -148,6 +148,43 @@
       }
     ]
   },
+  {
+    "testName": "tests the retry strategy on 5xx",
+    "autoCreateClient": false,
+    "steps": [
+      {
+        "type": "createClient",
+        "parameters": {
+          "appId": "test-app-id",
+          "apiKey": "test-api-key",
+          "customHosts": [
+            {
+              "port": 6671
+            },
+            {
+              "port": 6672
+            },
+            {
+              "port": 6673
+            }
+          ]
+        }
+      },
+      {
+        "type": "method",
+        "method": "customPost",
+        "parameters": {
+          "path": "1/test/error/${{language}}"
+        },
+        "expected": {
+          "type": "response",
+          "match": {
+            "status": "ok"
+          }
+        }
+      }
+    ]
+  },
   {
     "testName": "test the compression strategy",
     "autoCreateClient": false,