feat: implement bulk processing feature to support multiple datasets in TEE session

gradle.properties

@@ -1,8 +1,8 @@
 # x-release-please-start-version
 version=9.1.0
 # x-release-please-end
-iexecCommonVersion=9.0.0
-iexecCommonsPocoVersion=5.0.0
+iexecCommonsPocoVersion=5.2.0
+iexecCommonVersion=9.1.0
 
 nexusUser
 nexusPassword

src/main/java/com/iexec/sms/tee/TeeController.java

@@ -209,8 +209,8 @@ public ResponseEntity<ApiResponseBody<TeeSessionGenerationResponse, TeeSessionGe
         String taskId = workerpoolAuthorization.getChainTaskId();
         workerAddress = Keys.toChecksumAddress(workerAddress);
         String attestingEnclave = workerpoolAuthorization.getEnclaveChallenge();
-        log.info("TEE session request [taskId:{}, workerAddress:{}]",
-                taskId, workerAddress);
+        log.info("TEE session request [taskId:{}, dealId:{}, taskIndex:{}, workerAddress:{}]",
+                taskId, workerpoolAuthorization.getDealId(), workerpoolAuthorization.getTaskIndex(), workerAddress);
         try {
             TeeSessionGenerationResponse teeSessionGenerationResponse = teeSessionService
                     .generateTeeSession(taskId, workerAddress, attestingEnclave);

src/main/java/com/iexec/sms/tee/bulk/IpfsClient.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2025 IEXEC BLOCKCHAIN TECH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.iexec.sms.tee.bulk;
+
+import com.iexec.commons.poco.order.DatasetOrder;
+import feign.Param;
+import feign.RequestLine;
+
+import java.util.List;
+
+public interface IpfsClient {
+    @RequestLine("GET /ipfs/{cid}")
+    List<String> readBulkCid(@Param("cid") final String cid);
+
+    @RequestLine("GET /ipfs/{cid}")
+    List<DatasetOrder> readOrders(@Param("cid") final String cid);
+}

src/main/java/com/iexec/sms/tee/session/base/SecretSessionBaseService.java

@@ -16,16 +16,23 @@
 
 package com.iexec.sms.tee.session.base;
 
+import com.iexec.common.utils.FeignBuilder;
 import com.iexec.common.utils.IexecEnvUtils;
 import com.iexec.common.utils.IexecFileHelper;
+import com.iexec.commons.poco.chain.ChainDataset;
 import com.iexec.commons.poco.chain.DealParams;
+import com.iexec.commons.poco.order.DatasetOrder;
+import com.iexec.commons.poco.security.Signature;
 import com.iexec.commons.poco.task.TaskDescription;
 import com.iexec.commons.poco.tee.TeeEnclaveConfiguration;
+import com.iexec.commons.poco.utils.SignatureUtils;
+import com.iexec.sms.chain.IexecHubService;
 import com.iexec.sms.secret.compute.*;
 import com.iexec.sms.secret.web2.Web2Secret;
 import com.iexec.sms.secret.web2.Web2SecretHeader;
 import com.iexec.sms.secret.web2.Web2SecretService;
 import com.iexec.sms.secret.web3.Web3SecretService;
+import com.iexec.sms.tee.bulk.IpfsClient;
 import com.iexec.sms.tee.challenge.EthereumCredentials;
 import com.iexec.sms.tee.challenge.TeeChallenge;
 import com.iexec.sms.tee.challenge.TeeChallengeService;
@@ -35,10 +42,13 @@
 import com.iexec.sms.tee.session.generic.TeeSessionRequest;
 import com.iexec.sms.tee.session.gramine.GramineSessionMakerService;
 import com.iexec.sms.tee.session.scone.SconeSessionMakerService;
+import feign.Logger;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
+import java.math.BigInteger;
 import java.util.*;
 
 import static com.iexec.common.worker.tee.TeeSessionEnvironmentVariable.*;
@@ -57,23 +67,33 @@
 public class SecretSessionBaseService {
 
     static final String EMPTY_STRING_VALUE = "";
+    static final String BULK_DATASET_PREFIX = "BULK_DATASET_";
+    static final BigInteger BULK_DATASET_VOLUME = BigInteger.TWO.pow(53).subtract(BigInteger.ONE);
     static final String IEXEC_APP_DEVELOPER_SECRET_PREFIX = "IEXEC_APP_DEVELOPER_SECRET_";
     static final String IEXEC_REQUESTER_SECRET_PREFIX = "IEXEC_REQUESTER_SECRET_";
 
+    private final IpfsClient ipfsClient;
+    private final IexecHubService iexecHubService;
     private final Web3SecretService web3SecretService;
     private final Web2SecretService web2SecretService;
     private final TeeChallengeService teeChallengeService;
     private final TeeTaskComputeSecretService teeTaskComputeSecretService;
 
     public SecretSessionBaseService(
+            @Value("${ipfs-gateway}") final String ipfsGatewayUrl,
+            final IexecHubService iexecHubService,
             final Web3SecretService web3SecretService,
             final Web2SecretService web2SecretService,
             final TeeChallengeService teeChallengeService,
             final TeeTaskComputeSecretService teeTaskComputeSecretService) {
+        this.iexecHubService = iexecHubService;
         this.web3SecretService = web3SecretService;
         this.web2SecretService = web2SecretService;
         this.teeChallengeService = teeChallengeService;
         this.teeTaskComputeSecretService = teeTaskComputeSecretService;
+        this.ipfsClient = StringUtils.isEmpty(ipfsGatewayUrl)
+                ? null
+                : FeignBuilder.createBuilder(Logger.Level.BASIC).target(IpfsClient.class, ipfsGatewayUrl);
     }
 
     /**
@@ -99,8 +119,7 @@ public SecretSessionBase getSecretsTokens(final TeeSessionRequest request) throw
         final TaskDescription taskDescription = request.getTaskDescription();
         final Map<String, String> signTokens = getSignTokens(request);
         // pre-compute
-        final boolean isPreComputeRequired = taskDescription.containsDataset() || taskDescription.containsInputFiles();
-        if (isPreComputeRequired) {
+        if (taskDescription.requiresPreCompute()) {
             sessionBase.preCompute(getPreComputeTokens(request, signTokens));
         }
         // app
@@ -151,6 +170,66 @@ Map<String, String> getSignTokens(final TeeSessionRequest request) throws TeeSes
 
     // region pre-compute
 
+    private List<DatasetOrder> fetchDatasetOrders(final TaskDescription taskDescription) {
+        final String bulkCid = taskDescription.getDealParams().getBulkCid();
+        log.info("bulk [chainTaskId:{}, botIndex:{}, cid:{}]",
+                taskDescription.getChainTaskId(), taskDescription.getBotIndex(), bulkCid);
+        try {
+            // can be optimized and cached with final deadline
+            final List<String> datasets = ipfsClient.readBulkCid(bulkCid);
+            final int bulkIdx = taskDescription.getBotIndex() - taskDescription.getBotFirstIndex();
+            final List<DatasetOrder> tempList = ipfsClient.readOrders(datasets.get(bulkIdx));
+            return List.copyOf(tempList);
+        } catch (Exception e) {
+            log.error("Error during bulk computation", e);
+            return List.of();
+        }
+    }
+
+    private Map<String, Object> getBulkDatasetTokens(final int index,
+                                                     final TaskDescription taskDescription,
+                                                     final DatasetOrder datasetOrder) {
+        final String prefix = BULK_DATASET_PREFIX + (index + 1);
+        final ChainDataset dataset = iexecHubService.getChainDataset(datasetOrder.getDataset()).orElse(null);
+        if (isBulkDatasetOrderValid(taskDescription, datasetOrder) && dataset != null) {
+            final String datasetKey = web3SecretService.getDecryptedValue(datasetOrder.getDataset()).orElse("");
+            return Map.of(
+                    prefix + "_URL", dataset.getMultiaddr(),
+                    prefix + "_CHECKSUM", dataset.getChecksum(),
+                    prefix + "_KEY", datasetKey,
+                    prefix + "_FILENAME", datasetOrder.getDataset()
+            );
+        } else {
+            return Map.of(
+                    prefix + "_URL", "",
+                    prefix + "_CHECKSUM", "",
+                    prefix + "_KEY", "",
+                    prefix + "_FILENAME", datasetOrder.getDataset()
+            );
+        }
+    }
+
+    boolean isBulkDatasetOrderValid(final TaskDescription taskDescription, final DatasetOrder datasetOrder) {
+        try {
+            final Signature signature = new Signature(datasetOrder.getSign());
+            final String orderHash = datasetOrder.computeHash(iexecHubService.getOrdersDomain());
+            final String owner = iexecHubService.getOwner(datasetOrder.getDataset());
+            final boolean isSignedByOwner = SignatureUtils.doesSignatureMatchesAddress(
+                    signature.getR(), signature.getS(), orderHash, owner);
+            final BigInteger consumedVolume = iexecHubService.viewConsumed(orderHash);
+            final boolean isVolumeValid = BULK_DATASET_VOLUME.equals(datasetOrder.getVolume());
+            final boolean isOrderNotFullyConsumed = !BULK_DATASET_VOLUME.equals(consumedVolume);
+            final boolean isTagValid = taskDescription.getTag().equals(datasetOrder.getTag());
+            log.info("check bulk dataset order [chainTaskId:{}, dataset:{}, owner:{}, isSignedByOwner:{}, isVolumeValid:{}, isOrderNotFullyConsumed:{}, isTagValid:{}]",
+                    taskDescription.getChainTaskId(), datasetOrder.getDataset(), owner, isSignedByOwner, isVolumeValid, isOrderNotFullyConsumed, isTagValid);
+            return isSignedByOwner && isVolumeValid && isOrderNotFullyConsumed && isTagValid;
+        } catch (Exception e) {
+            log.error("failed to perform all checks on dataset [chainTaskId:{}, dataset:{}]",
+                    taskDescription.getChainTaskId(), datasetOrder.getDataset());
+            return false;
+        }
+    }
+
     /**
      * Get tokens to be injected in the pre-compute enclave.
      *
@@ -170,6 +249,16 @@ SecretEnclaveBase getPreComputeTokens(final TeeSessionRequest request, final Map
         // `IS_DATASET_REQUIRED` still meaningful?
         tokens.put(IS_DATASET_REQUIRED.name(), taskDescription.containsDataset());
 
+        if (taskDescription.isBulkRequest()) {
+            final List<DatasetOrder> orders = fetchDatasetOrders(taskDescription);
+            tokens.put(BULK_SIZE.name(), orders.size());
+            for (int i = 0; i < orders.size(); i++) {
+                final DatasetOrder order = orders.get(i);
+                log.info("order found {}", order);
+                tokens.putAll(getBulkDatasetTokens(i, taskDescription, order));
+            }
+        }
+
         final List<String> trustedEnv = new ArrayList<>();
         if (taskDescription.containsDataset()) {
             final String datasetKey = web3SecretService
@@ -243,6 +332,17 @@ SecretEnclaveBase getAppTokens(final TeeSessionRequest request) throws TeeSessio
         tokens.putAll(computeSecrets);
         // trusted env variables (not confidential)
         tokens.putAll(IexecEnvUtils.getComputeStageEnvMap(taskDescription));
+
+        if (taskDescription.isBulkRequest()) {
+            final List<String> addresses = fetchDatasetOrders(taskDescription).stream()
+                    .map(DatasetOrder::getDataset)
+                    .toList();
+            tokens.put(BULK_SIZE.name(), addresses.size());
+            for (int i = 0; i < addresses.size(); i++) {
+                tokens.put(BULK_DATASET_PREFIX + (i + 1) + "_FILENAME", addresses.get(i));
+            }
+        }
+
         return enclaveBase
                 .environment(tokens)
                 .build();

src/main/resources/application.yml

@@ -42,6 +42,8 @@ chain:
   gas-price-multiplier: ${IEXEC_GAS_PRICE_MULTIPLIER:1.0} # txs will be sent with networkGasPrice*gasPriceMultiplier, 4.0 means superfast
   gas-price-cap: ${IEXEC_GAS_PRICE_CAP:22000000000} #in Wei, will be used for txs if networkGasPrice*gasPriceMultiplier > gasPriceCap
 
+ipfs-gateway: https://ipfs.iex.ec
+
 metrics:
   storage:
     refresh-interval: ${IEXEC_SMS_METRICS_STORAGE_REFRESH_INTERVAL:30}  # In seconds

src/test/java/com/iexec/sms/tee/session/TeeSessionTestUtils.java

@@ -18,8 +18,10 @@
 
 import com.iexec.common.utils.FileHashUtils;
 import com.iexec.commons.poco.chain.DealParams;
+import com.iexec.commons.poco.order.OrderTag;
 import com.iexec.commons.poco.task.TaskDescription;
 import com.iexec.commons.poco.tee.TeeEnclaveConfiguration;
+import com.iexec.commons.poco.utils.BytesUtils;
 import com.iexec.sms.api.config.SconeServicesProperties;
 import com.iexec.sms.api.config.TeeAppProperties;
 import com.iexec.sms.api.config.TeeServicesProperties;
@@ -36,6 +38,7 @@
 
 @Slf4j
 public class TeeSessionTestUtils {
+    public static final String TEST_DEAL_ID = "dealId";
     public static final String TASK_ID = "taskId";
     public static final String SESSION_ID = "sessionId";
     public static final String WORKER_ADDRESS = "workerAddress";
@@ -134,27 +137,35 @@ public static DealParams.DealParamsBuilder createDealParams() {
                 .iexecSecrets(Map.of("1", REQUESTER_SECRET_KEY_1, "2", REQUESTER_SECRET_KEY_2));
     }
 
-    public static TaskDescription.TaskDescriptionBuilder createTaskDescription(TeeEnclaveConfiguration enclaveConfig) {
+    public static TaskDescription.TaskDescriptionBuilder createTaskDescription(final DealParams dealParams,
+                                                                               final TeeEnclaveConfiguration enclaveConfig) {
         final String appAddress = createEthereumAddress();
         final String requesterAddress = createEthereumAddress();
         final String beneficiaryAddress = createEthereumAddress();
         final String workerpoolAddress = createEthereumAddress();
-        final DealParams dealParams = createDealParams().build();
         return TaskDescription.builder()
-                .workerpoolOwner(workerpoolAddress)
+                .chainDealId(TEST_DEAL_ID)
                 .chainTaskId(TASK_ID)
+                .workerpoolOwner(workerpoolAddress)
                 .appUri(APP_URI)
                 .appAddress(appAddress)
                 .appEnclaveConfiguration(enclaveConfig)
-                .datasetAddress(DATASET_ADDRESS)
-                .datasetUri(DATASET_URL)
-                .datasetChecksum(DATASET_CHECKSUM)
+                .datasetAddress(BytesUtils.EMPTY_ADDRESS)
+                .tag(OrderTag.TEE_SCONE.getValue())
                 .requester(requesterAddress)
                 .beneficiary(beneficiaryAddress)
                 .dealParams(dealParams)
                 .botSize(1)
                 .botFirstIndex(0)
                 .botIndex(0);
     }
+
+    public static TaskDescription.TaskDescriptionBuilder createTaskDescriptionWithDataset(final DealParams dealParams,
+                                                                                          final TeeEnclaveConfiguration enclaveConfiguration) {
+        return createTaskDescription(dealParams, enclaveConfiguration)
+                .datasetAddress(DATASET_ADDRESS)
+                .datasetUri(DATASET_URL)
+                .datasetChecksum(DATASET_CHECKSUM);
+    }
     //endregion
 }