feat: add waitForResult option to processBulkRequest

Author: PierreJeanjacquot

packages/sdk/src/lib/dataProtectorCore/IExecDataProtectorCore.ts

@@ -100,9 +100,9 @@ class IExecDataProtectorCore extends IExecDataProtectorModule {
     });
   }
 
-  async processBulkRequest(
-    args: ProcessBulkRequestParams
-  ): Promise<ProcessBulkRequestResponse> {
+  async processBulkRequest<Params extends ProcessBulkRequestParams>(
+    args: Params
+  ): Promise<ProcessBulkRequestResponse<Params>> {
     await this.init();
     await isValidProvider(this.iexec);
     return processBulkRequest({

packages/sdk/src/lib/dataProtectorCore/processBulkRequest.ts

@@ -25,9 +25,13 @@ import {
   OnStatusUpdateFn,
   ProcessBulkRequestParams,
   ProcessBulkRequestResponse,
+  ProcessBulkRequestResponseBase,
+  ProcessBulkRequestResponseWithResult,
   ProcessBulkRequestStatuses,
 } from '../types/index.js';
 import { IExecConsumer, VoucherInfo } from '../types/internalTypes.js';
+import { getResultFromCompletedTask } from './getResultFromCompletedTask.js';
+import { waitForTaskCompletion } from './waitForTaskCompletion.js';
 
 export type ProcessBulkRequest = typeof processBulkRequest;
 
@@ -38,18 +42,22 @@ const waitForRetry = (ms: number): Promise<void> => {
   });
 };
 
-export const processBulkRequest = async ({
+export const processBulkRequest = async <
+  Params extends ProcessBulkRequestParams
+>({
   iexec = throwIfMissing(),
   defaultWorkerpool,
   bulkRequest,
   workerpool,
   useVoucher = false,
   voucherOwner,
+  path,
   pemPrivateKey,
+  waitForResult = false,
   onStatusUpdate = () => {},
-}: IExecConsumer &
-  DefaultWorkerpoolConsumer &
-  ProcessBulkRequestParams): Promise<ProcessBulkRequestResponse> => {
+}: IExecConsumer & DefaultWorkerpoolConsumer & Params): Promise<
+  ProcessBulkRequestResponse<Params>
+> => {
   const vRequestorder = bulkRequestSchema()
     .label('bulkRequest')
     .required()
@@ -64,6 +72,10 @@ export const processBulkRequest = async ({
   const vVoucherOwner = addressOrEnsSchema()
     .label('voucherOwner')
     .validateSync(voucherOwner);
+  const vWaitForResult = booleanSchema()
+    .label('waitForResult')
+    .validateSync(waitForResult);
+  const vPath = stringSchema().label('path').validateSync(path);
   const vPemPrivateKey = stringSchema()
     .label('pemPrivateKey')
     .validateSync(pemPrivateKey);
@@ -72,17 +84,22 @@ export const processBulkRequest = async ({
       OnStatusUpdateFn<ProcessBulkRequestStatuses>
     >(onStatusUpdate);
 
-  // Validate that pemPrivateKey is provided if iexec_result_encryption is true
-  if (
+  const iexecResultEncryption =
     // JSON parse safe thanks to bulkRequestSchema validation
-    JSON.parse(vRequestorder.params)?.iexec_result_encryption === true &&
-    !vPemPrivateKey
-  ) {
+    JSON.parse(vRequestorder.params)?.iexec_result_encryption === true;
+  // Validate that pemPrivateKey is provided if iexec_result_encryption is true
+  if (vWaitForResult && iexecResultEncryption && !vPemPrivateKey) {
     throw new ValidationError(
       'Missing pemPrivateKey required for result decryption'
     );
   }
 
+  if (vWaitForResult && !iexecResultEncryption && vPemPrivateKey) {
+    throw new ValidationError(
+      'pemPrivateKey is passed but result encryption is not enabled in bulkRequest this is likely an error when preparing the bulk request'
+    );
+  }
+
   try {
     let userVoucher: VoucherInfo | undefined;
     if (vUseVoucher) {
@@ -241,11 +258,7 @@ export const processBulkRequest = async ({
       { pageSize: Math.max(volume, 10) } // Fetch all deals (min page size 10)
     );
 
-    const tasks: Array<{
-      taskId: string;
-      dealId: string;
-      bulkIndex: number;
-    }> = [];
+    const tasks: ProcessBulkRequestResponseBase['tasks'] = [];
 
     for (const deal of deals) {
       const dealTasks = await Promise.all(
@@ -259,6 +272,7 @@ export const processBulkRequest = async ({
       );
       tasks.push(...dealTasks);
     }
+    tasks.sort((a, b) => a.bulkIndex - b.bulkIndex);
 
     vOnStatusUpdate({
       title: 'MATCH_ORDERS_LOOP',
@@ -269,8 +283,74 @@ export const processBulkRequest = async ({
       },
     });
 
+    if (!vWaitForResult) {
+      return {
+        tasks,
+      } as ProcessBulkRequestResponse<Params>;
+    }
+
+    const tasksWithResults =
+      tasks as ProcessBulkRequestResponseWithResult['tasks'];
+
+    await Promise.all(
+      tasksWithResults.map(async (task) => {
+        try {
+          vOnStatusUpdate({
+            title: 'PROCESS_BULK_SLICE',
+            isDone: false,
+            payload: task,
+          });
+          vOnStatusUpdate({
+            title: 'CONSUME_TASK',
+            isDone: false,
+            payload: task,
+          });
+          const { status, success } = await waitForTaskCompletion({
+            iexec,
+            taskId: task.taskId,
+            dealId: task.dealId,
+          });
+          task.status = status;
+          task.success = success;
+          vOnStatusUpdate({
+            title: 'CONSUME_TASK',
+            isDone: true,
+            payload: task,
+          });
+          if (!success) {
+            throw new Error(`Task ended with status: ${status}`);
+          }
+          const { result } = await getResultFromCompletedTask({
+            iexec,
+            taskId: task.taskId,
+            path: vPath,
+            pemPrivateKey: vPemPrivateKey,
+            onStatusUpdate: (update) => {
+              vOnStatusUpdate({
+                ...update,
+                payload: { ...update.payload, task },
+              });
+            },
+          });
+          task.result = result;
+          vOnStatusUpdate({
+            title: 'PROCESS_BULK_SLICE',
+            isDone: true,
+            payload: task,
+          });
+        } catch (error) {
+          task.error = error as Error;
+          vOnStatusUpdate({
+            title: 'PROCESS_BULK_SLICE',
+            isDone: true,
+            payload: task,
+          });
+        }
+      })
+    );
+
     return {
-      tasks: tasks.sort((a, b) => a.bulkIndex - b.bulkIndex),
+      tasks: tasksWithResults,
     };
   } catch (error) {
     console.error('[processBulkRequest] ERROR', error);

packages/sdk/src/lib/types/coreTypes.ts

@@ -233,8 +233,10 @@ export type WaitForTaskCompletionParams = {
   onStatusUpdate?: OnStatusUpdateFn<WaitForTaskCompletionStatuses>;
 };
 
+export type TaskStatus = 'COMPLETED' | 'FAILED' | 'TIMEOUT';
+
 export type WaitForTaskCompletionResponse = {
-  status: 'COMPLETED' | 'FAILED' | 'TIMEOUT';
+  status: TaskStatus;
   success: boolean;
 };
 
@@ -354,6 +356,8 @@ export type ProcessProtectedDataParams = {
 
   /**
    * The file name of the desired file in the returned ZIP file.
+   *
+   * Ignored if `waitForResult` is `false`
    */
   path?: string;
 
@@ -390,7 +394,7 @@ export type ProcessProtectedDataParams = {
 
   /**
    * Enable result encryption for the processed data.
-   * @default = false
+   * @default false
    */
   encryptResult?: boolean;
 
@@ -402,7 +406,7 @@ export type ProcessProtectedDataParams = {
 
   /**
    * Whether to wait for the result of the processing or not.
-   * @default = true
+   * @default true
    */
   waitForResult?: boolean;
 
@@ -535,7 +539,8 @@ export type ProcessBulkRequestStatuses =
   | 'REQUEST_TO_PROCESS_BULK_DATA'
   | 'CONSUME_TASK'
   | 'CONSUME_RESULT_DOWNLOAD'
-  | 'CONSUME_RESULT_DECRYPT';
+  | 'CONSUME_RESULT_DECRYPT'
+  | 'PROCESS_BULK_SLICE';
 
 export type ProcessBulkRequestParams = {
   /**
@@ -545,6 +550,8 @@ export type ProcessBulkRequestParams = {
 
   /**
    * Path to the result file in the app's output
+   *
+   * Ignored if `waitForResult` is `false`
    */
   path?: string;
 
@@ -565,20 +572,43 @@ export type ProcessBulkRequestParams = {
 
   /**
    * Private key in PEM format for result decryption.
-   * required if bulkRequest use results encryption.
+   *
+   * Required if `bulkRequest` use results encryption and `waitForResult` is `true`.
    */
   pemPrivateKey?: string;
 
+  /**
+   * Whether to wait for the result of the bulk request.
+   * @default false
+   */
+  waitForResult?: boolean;
+
   /**
    * Callback function that will get called at each step of the process
    */
   onStatusUpdate?: OnStatusUpdateFn<ProcessBulkRequestStatuses>;
 };
 
-export type ProcessBulkRequestResponse = {
+export type ProcessBulkRequestResponse<T> = T extends { waitForResult: true }
+  ? ProcessBulkRequestResponseWithResult
+  : ProcessBulkRequestResponseBase;
+
+export type ProcessBulkRequestResponseBase = {
+  tasks: Array<{
+    taskId: string;
+    dealId: string;
+    bulkIndex: number;
+  }>;
+};
+
+export type ProcessBulkRequestResponseWithResult = {
   tasks: Array<{
     taskId: string;
     dealId: string;
     bulkIndex: number;
+    success: boolean;
+    status: TaskStatus;
+    result?: ArrayBuffer;
+    error?: Error;
   }>;
 };