fix(clients): reduce chances of Push rate limiting (#5153)

Author: shortcuts

clients/algoliasearch-client-java/.java-version

@@ -1 +1 @@
-21.0.8
+21.0.6

clients/algoliasearch-client-php/lib/FormDataProcessor.php

@@ -12,9 +12,9 @@
  */
 
 /**
- * Search API.
+ * Ingestion API.
  *
- * The Algolia Search API lets you search, configure, and manage your indices and records.  ## Client libraries  Use Algolia's API clients and libraries to reliably integrate Algolia's APIs with your apps. The official API clients are covered by Algolia's [Service Level Agreement](https://www.algolia.com/policies/sla/).  See: [Algolia's ecosystem](https://www.algolia.com/doc/guides/getting-started/how-algolia-works/in-depth/ecosystem/)  ## Base URLs  The base URLs for requests to the Search API are:  - `https://{APPLICATION_ID}.algolia.net` - `https://{APPLICATION_ID}-dsn.algolia.net`.   If your subscription includes a [Distributed Search Network](https://dashboard.algolia.com/infra),   this ensures that requests are sent to servers closest to users.  Both URLs provide high availability by distributing requests with load balancing.  **All requests must use HTTPS.**  ## Retry strategy  To guarantee high availability, implement a retry strategy for all API requests using the URLs of your servers as fallbacks:  - `https://{APPLICATION_ID}-1.algolianet.com` - `https://{APPLICATION_ID}-2.algolianet.com` - `https://{APPLICATION_ID}-3.algolianet.com`  These URLs use a different DNS provider than the primary URLs. You should randomize this list to ensure an even load across the three servers.  All Algolia API clients implement this retry strategy.  ## Authentication  To authenticate your API requests, add these headers:  - `x-algolia-application-id`. Your Algolia application ID. - `x-algolia-api-key`. An API key with the necessary permissions to make the request.   The required access control list (ACL) to make a request is listed in each endpoint's reference.  You can find your application ID and API key in the [Algolia dashboard](https://dashboard.algolia.com/account).  ## Request format  Depending on the endpoint, request bodies are either JSON objects or arrays of JSON objects,  ## Parameters  Parameters are passed as query parameters for GET and DELETE requests, and in the request body for POST and PUT requests.  Query parameters must be [URL-encoded](https://developer.mozilla.org/en-US/docs/Glossary/Percent-encoding). Non-ASCII characters must be UTF-8 encoded. Plus characters (`+`) are interpreted as spaces. Arrays as query parameters must be one of:  - A comma-separated string: `attributesToRetrieve=title,description` - A URL-encoded JSON array: `attributesToRetrieve=%5B%22title%22,%22description%22%D`  ## Response status and errors  The Search API returns JSON responses. Since JSON doesn't guarantee any specific ordering, don't rely on the order of attributes in the API response.  Successful responses return a `2xx` status. Client errors return a `4xx` status. Server errors are indicated by a `5xx` status. Error responses have a `message` property with more information.  ## Version  The current version of the Search API is version 1, as indicated by the `/1/` in each endpoint's URL.
+ * The Ingestion API lets you connect third-party services and platforms with Algolia and schedule tasks to ingest your data. The Ingestion API powers the no-code [data connectors](https://dashboard.algolia.com/connectors).  ## Base URLs  The base URLs for requests to the Ingestion API are:  - `https://data.us.algolia.com` - `https://data.eu.algolia.com`  Use the URL that matches your [analytics region](https://dashboard.algolia.com/account/infrastructure/analytics).  **All requests must use HTTPS.**  ## Authentication  To authenticate your API requests, add these headers:  - `x-algolia-application-id`. Your Algolia application ID. - `x-algolia-api-key`. An API key with the necessary permissions to make the request.   The required access control list (ACL) to make a request is listed in each endpoint's reference.  You can find your application ID and API key in the [Algolia dashboard](https://dashboard.algolia.com/account).  ## Request format  Request bodies must be JSON objects.  ## Response status and errors  Response bodies are JSON objects.  Successful responses return a `2xx` status. Client errors return a `4xx` status. Server errors are indicated by a `5xx` status. Error responses have a `message` property with more information.  ## Version  The current version of the Ingestion API is version 1, as indicated by the `/1/` in each endpoint's URL.
  *
  * The version of the OpenAPI document: 1.0.0
  * Generated by: https://openapi-generator.tech
@@ -29,7 +29,7 @@
 
 namespace Algolia\AlgoliaSearch;
 
-use Algolia\AlgoliaSearch\Model\Search\ModelInterface;
+use Algolia\AlgoliaSearch\Model\Ingestion\ModelInterface;
 use DateTime;
 use GuzzleHttp\Psr7\Utils;
 use Psr\Http\Message\StreamInterface;

playground/javascript/node/algoliasearch.ts

@@ -152,7 +152,7 @@ async function testAlgoliasearchBridgeIngestion() {
   console.log('replaceAllObjectsWithTransformation', await client.replaceAllObjectsWithTransformation({
     indexName: 'boyd',
     objects: [{ objectID: 'foo', data: { baz: 'baz', win: 42 } }, { objectID: 'bar', data: { baz: 'baz', win: 24 } }],
-    batchSize: 2
+    batchSize: 1
   }));
 }
 

scripts/cts/testServer/replaceAllObjectsWithTransformation.ts

@@ -10,7 +10,9 @@ const raowtState: Record<
   string,
   {
     copyCount: number;
+    deleteCount: number;
     pushCount: number;
+    getEventCount: number;
     tmpIndexName: string;
     waitTaskCount: number;
     waitingForFinalWaitTask: boolean;
@@ -33,6 +35,28 @@ function addRoutes(app: Express): void {
     }),
   );
 
+  app.delete('/1/indexes/:indexName', (req, res) => {
+    expect(req.params.indexName).to.match(/^cts_e2e_replace_all_objects_with_transformation_(.*)$/);
+
+    const lang = req.params.indexName.replace('cts_e2e_replace_all_objects_with_transformation_', '');
+    if (!raowtState[lang] || raowtState[lang].successful) {
+      raowtState[lang] = {
+        copyCount: 0,
+        pushCount: 0,
+        getEventCount: 0,
+        deleteCount: 1,
+        waitTaskCount: 0,
+        tmpIndexName: req.params.indexName,
+        waitingForFinalWaitTask: false,
+        successful: false,
+      };
+    } else {
+      raowtState[lang].deleteCount++;
+    }
+
+    res.json({ taskID: 123 + raowtState[lang].copyCount, deletedAt: '2021-01-01T00:00:00.000Z' });
+  });
+
   app.post('/1/indexes/:indexName/operation', (req, res) => {
     expect(req.params.indexName).to.match(/^cts_e2e_replace_all_objects_with_transformation_(.*)$/);
 
@@ -47,6 +71,8 @@ function addRoutes(app: Express): void {
           raowtState[lang] = {
             copyCount: 1,
             pushCount: 0,
+            getEventCount: 0,
+            deleteCount: 0,
             waitTaskCount: 0,
             tmpIndexName: req.body.destination,
             waitingForFinalWaitTask: false,
@@ -62,9 +88,12 @@ 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: 10,
+          pushCount: 4,
+          getEventCount: 4,
+          deleteCount: 0,
           waitTaskCount: 2,
           tmpIndexName: req.params.indexName,
           waitingForFinalWaitTask: false,
@@ -94,21 +123,25 @@ function addRoutes(app: Express): void {
     expect(req.body.action === 'addObject').to.equal(true);
     expect(req.query.referenceIndexName === `cts_e2e_replace_all_objects_with_transformation_${lang}`).to.equal(true);
 
-    raowtState[lang].pushCount += req.body.records.length;
+    raowtState[lang].pushCount++;
 
     res.json({
-      runID: 'b1b7a982-524c-40d2-bb7f-48aab075abda',
+      runID: `b1b7a982-524c-40d2-bb7f-48aab075abda_${lang}`,
       eventID: `113b2068-6337-4c85-b5c2-e7b213d8292${raowtState[lang].pushCount}`,
       message: 'OK',
       createdAt: '2022-05-12T06:24:30.049Z',
     });
   });
 
   app.get('/1/runs/:runID/events/:eventID', (req, res) => {
+    const lang = req.params.runID.match(/^b1b7a982-524c-40d2-bb7f-48aab075abda_(.*)$/)?.[1] as string;
+
+    raowtState[lang].getEventCount++;
+
     res.json({
       status: 'succeeded',
-      eventID: '113b2068-6337-4c85-b5c2-e7b213d82921',
-      runID: 'b1b7a982-524c-40d2-bb7f-48aab075abda',
+      eventID: req.params.eventID,
+      runID: req.params.runID,
       type: 'fetch',
       batchSize: 1,
       publishedAt: '2022-05-12T06:24:30.049Z',
@@ -123,10 +156,18 @@ function addRoutes(app: Express): void {
 
     raowtState[lang].waitTaskCount++;
     if (raowtState[lang].waitingForFinalWaitTask) {
-      expect(req.params.taskID).to.equal('777');
-      expect(raowtState[lang].waitTaskCount).to.equal(3);
-
       raowtState[lang].successful = true;
+      expect(req.params.taskID).to.equal('777');
+      expect(raowtState[lang]).to.deep.equal({
+        copyCount: 2,
+        pushCount: 4,
+        getEventCount: 4,
+        deleteCount: 0,
+        waitTaskCount: 3,
+        tmpIndexName: req.params.indexName,
+        waitingForFinalWaitTask: true,
+        successful: true,
+      });
     }
 
     res.json({ status: 'published', updatedAt: '2021-01-01T00:00:00.000Z' });

templates/go/ingestion_helpers.mustache

@@ -16,6 +16,12 @@ func (c *APIClient) ChunkedPush(indexName string, objects []map[string]any, acti
 		batchSize:    1000,
 	}
 
+	offset := 0
+	waitBatchSize := conf.batchSize / 10
+	if waitBatchSize < 1 {
+		waitBatchSize = conf.batchSize
+	}
+
 	for _, opt := range opts {
 		opt.apply(&conf)
 	}
@@ -58,31 +64,42 @@ func (c *APIClient) ChunkedPush(indexName string, objects []map[string]any, acti
 			responses = append(responses, *resp)
 			records = make([]map[string]any, 0, len(objects)%conf.batchSize)
 		}
-	}
 
-	if conf.waitForTasks {
-		for _, resp := range responses {
-      _, err := CreateIterable( //nolint:wrapcheck
-				func(*Event, error) (*Event, error) {
-          if resp.EventID == nil {
-				    return nil, reportError("received unexpected response from the push endpoint, eventID must not be undefined")
-          }
-
-					return c.GetEvent(c.NewApiGetEventRequest(resp.RunID, *resp.EventID))
-				},
-				func(response *Event, err error) (bool, error) {
-					var apiErr *APIError
-					if errors.As(err, &apiErr) {
-						return apiErr.Status != 404, nil
-					}
-
-					return true, err
-				},
-				WithTimeout(func(count int) time.Duration { return time.Duration(min(500*count, 5000)) * time.Millisecond }), WithMaxRetries(50),
-			)
-			if err != nil {
-				return nil, err
+
+		if conf.waitForTasks && len(responses) > 0 && (len(responses)%waitBatchSize == 0 || i == len(objects)-1) {
+			var waitableResponses []WatchResponse
+
+			if len(responses) > offset+waitBatchSize {
+				waitableResponses = responses[offset:waitBatchSize]
+			} else {
+				waitableResponses = responses[offset:]
 			}
+
+			for _, resp := range waitableResponses {
+				_, err := CreateIterable(
+					func(*Event, error) (*Event, error) {
+						if resp.EventID == nil {
+							return nil, reportError("received unexpected response from the push endpoint, eventID must not be undefined")
+						}
+
+						return c.GetEvent(c.NewApiGetEventRequest(resp.RunID, *resp.EventID))
+					},
+					func(response *Event, err error) (bool, error) {
+						var apiErr *APIError
+						if errors.As(err, &apiErr) {
+							return apiErr.Status != 404, nil
+						}
+
+						return true, err
+					},
+					WithTimeout(func(count int) time.Duration { return time.Duration(min(500*count, 5000)) * time.Millisecond }), WithMaxRetries(50),
+				)
+				if err != nil {
+					return nil, err
+				}
+			}
+
+			offset += waitBatchSize
 		}
 	}
 

templates/java/api.mustache

@@ -20,6 +20,7 @@ import java.util.function.IntUnaryOperator;
 
 import java.time.Duration;
 import java.util.EnumSet;
+import java.util.Iterator;
 import java.util.Random;
 import java.util.Collections;
 import java.util.ArrayList;

templates/java/api_helpers.mustache

@@ -44,58 +44,62 @@ public <T> List<WatchResponse> chunkedPush(
 ) {
   List<WatchResponse> responses = new ArrayList<>();
   List<T> records = new ArrayList<>();
+  int offset = 0;
+  int waitBatchSize = batchSize / 10;
+  if (waitBatchSize < 1) {
+    waitBatchSize = batchSize;
+  }
 
-  for (T item : objects) {
-    if (records.size() == batchSize) {
-      WatchResponse watch =
-        this.push(
-            indexName,
-            new PushTaskPayload().setAction(action).setRecords(this.objectsToPushTaskRecords(records)),
-            waitForTasks,
-            referenceIndexName,
-            requestOptions
-          );
+  Iterator<T> it = objects.iterator();
+  T current = it.next();
+
+  while (true) {
+    records.add(current);
+
+    if (records.size() == batchSize || !it.hasNext()) {
+      WatchResponse watch = this.push(
+        indexName,
+        new PushTaskPayload().setAction(action).setRecords(this.objectsToPushTaskRecords(records)),
+        waitForTasks,
+        referenceIndexName,
+        requestOptions
+      );
       responses.add(watch);
       records.clear();
     }
 
-    records.add(item);
-  }
+    if (waitForTasks && responses.size() > 0 && (responses.size() % waitBatchSize == 0 || !it.hasNext())) {
+      responses
+        .subList(offset, Math.min(offset + waitBatchSize, responses.size()))
+        .forEach(response -> {
+          TaskUtils.retryUntil(
+            () -> {
+              try {
+                return this.getEvent(response.getRunID(), response.getEventID());
+              } catch (AlgoliaApiException e) {
+                if (e.getStatusCode() == 404) {
+                  return null;
+                }
+
+                throw e;
+              }
+            },
+            (Event resp) -> {
+              return resp != null;
+            },
+            50,
+            null
+          );
+        });
 
-  if (records.size() > 0) {
-    WatchResponse watch =
-      this.push(
-          indexName,
-          new PushTaskPayload().setAction(action).setRecords(this.objectsToPushTaskRecords(records)),
-          waitForTasks,
-          referenceIndexName,
-          requestOptions
-        );
-    responses.add(watch);
-  }
+      offset += waitBatchSize;
+    }
 
-  if (waitForTasks) {
-    responses.forEach(response -> {
-        TaskUtils.retryUntil(
-          () -> {
-            try {
-              return this.getEvent(response.getRunID(), response.getEventID());
-            } catch (AlgoliaApiException e) {
-              if (e.getStatusCode() == 404) {
-                return null;
-              }
+    if (!it.hasNext()) {
+      break;
+    }
 
-              throw e;
-            }
-          },
-          (Event resp) -> {
-            return resp != null;
-          },
-          50,
-          null
-        );
-      }
-    );
+    current = it.next();
   }
 
   return responses;