feat: implement bulk processing feature to support multiple datasets in TEE session (#310)

Author: jbern0rd

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/bulk/IpfsConfiguration.java

@@ -0,0 +1,43 @@
+/*
+ * 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.common.utils.FeignBuilder;
+import feign.Logger;
+import jakarta.validation.constraints.NotEmpty;
+import lombok.Value;
+import lombok.extern.slf4j.Slf4j;
+import org.hibernate.validator.constraints.URL;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.validation.annotation.Validated;
+
+@Slf4j
+@Value
+@Validated
+@ConfigurationProperties(prefix = "ipfs")
+public class IpfsConfiguration {
+    @URL
+    @NotEmpty
+    String gatewayUrl;
+
+    @Bean
+    IpfsClient ipfsClient() {
+        log.info("starting with ipfs located at {}", gatewayUrl);
+        return FeignBuilder.createBuilder(Logger.Level.BASIC).target(IpfsClient.class, gatewayUrl);
+    }
+}

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

@@ -18,14 +18,20 @@
 
 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;
@@ -39,6 +45,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;
 
+import java.math.BigInteger;
 import java.util.*;
 
 import static com.iexec.common.worker.tee.TeeSessionEnvironmentVariable.*;
@@ -57,23 +64,35 @@
 public class SecretSessionBaseService {
 
     static final String EMPTY_STRING_VALUE = "";
+    static final BigInteger BULK_DATASET_VOLUME = BigInteger.TWO.pow(53).subtract(BigInteger.ONE);
+    static final String BULK_DATASET_PREFIX = "BULK_DATASET_";
+    static final String BULK_DATASET_URL_SUFFIX = "_URL";
+    static final String BULK_DATASET_CHECKSUM_SUFFIX = "_CHECKSUM";
+    static final String BULK_DATASET_KEY_SUFFIX = "_KEY";
+    static final String BULK_DATASET_FILENAME_SUFFIX = "_FILENAME";
     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(
+            final IpfsClient ipfsClient,
+            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 = ipfsClient;
     }
 
     /**
@@ -99,8 +118,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 +169,71 @@ Map<String, String> getSignTokens(final TeeSessionRequest request) throws TeeSes
 
     // region pre-compute
 
+    /**
+     * Fetch dataset orders related to a bulk processing slice from IPFS
+     *
+     * @param taskDescription A task part of a bulk processing deal and corresponding to one of the slices
+     * @return The list of {@code DatasetOrder} found in the slice, or an empty list if any issue arises
+     */
+    private List<DatasetOrder> fetchDatasetOrders(final TaskDescription taskDescription) {
+        try {
+            final String bulkCid = taskDescription.getDealParams().getBulkCid();
+            final int bulkSliceIndex = taskDescription.getBotIndex() - taskDescription.getBotFirstIndex();
+            log.info("Fetching dataset orders for a bulk slice [chainTaskId:{}, bulkCid:{}, bulkSliceIndex:{}]",
+                    taskDescription.getChainTaskId(), bulkCid, bulkSliceIndex);
+            final List<String> bulkSlices = ipfsClient.readBulkCid(bulkCid);
+            final List<DatasetOrder> tempList = ipfsClient.readOrders(bulkSlices.get(bulkSliceIndex));
+            return List.copyOf(tempList);
+        } catch (final 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 + BULK_DATASET_URL_SUFFIX, dataset.getMultiaddr(),
+                    prefix + BULK_DATASET_CHECKSUM_SUFFIX, dataset.getChecksum(),
+                    prefix + BULK_DATASET_KEY_SUFFIX, datasetKey,
+                    prefix + BULK_DATASET_FILENAME_SUFFIX, datasetOrder.getDataset()
+            );
+        } else {
+            return Map.of(
+                    prefix + BULK_DATASET_URL_SUFFIX, EMPTY_STRING_VALUE,
+                    prefix + BULK_DATASET_CHECKSUM_SUFFIX, EMPTY_STRING_VALUE,
+                    prefix + BULK_DATASET_KEY_SUFFIX, EMPTY_STRING_VALUE,
+                    prefix + BULK_DATASET_FILENAME_SUFFIX, 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 +253,15 @@ 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);
+                tokens.putAll(getBulkDatasetTokens(i, taskDescription, order));
+            }
+        }
+
         final List<String> trustedEnv = new ArrayList<>();
         if (taskDescription.containsDataset()) {
             final String datasetKey = web3SecretService
@@ -243,6 +335,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) + BULK_DATASET_FILENAME_SUFFIX, addresses.get(i));
+            }
+        }
+
         return enclaveBase
                 .environment(tokens)
                 .build();

src/main/resources/application.yml

@@ -42,6 +42,9 @@ 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-url: https://ipfs.iex.ec
+
 metrics:
   storage:
     refresh-interval: ${IEXEC_SMS_METRICS_STORAGE_REFRESH_INTERVAL:30}  # In seconds

src/test/java/com/iexec/sms/tee/bulk/IpsConfigurationTests.java

@@ -0,0 +1,67 @@
+/*
+ * 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 jakarta.validation.ConstraintViolation;
+import jakarta.validation.Validation;
+import jakarta.validation.Validator;
+import jakarta.validation.ValidatorFactory;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class IpsConfigurationTests {
+    private Validator validator;
+
+    @BeforeEach
+    void setUp() {
+        try (final ValidatorFactory factory = Validation.buildDefaultValidatorFactory()) {
+            validator = factory.getValidator();
+        }
+    }
+
+    @ParameterizedTest
+    @NullSource
+    @ValueSource(strings = "")
+    void shouldNotBeValidWithEmptyValues(final String gatewayUrl) {
+        final IpfsConfiguration config = new IpfsConfiguration(gatewayUrl);
+        assertThat(validator.validate(config))
+                .isNotEmpty()
+                .extracting(ConstraintViolation::getMessage)
+                .containsExactly("must not be empty");
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {" ", "not-an-url"})
+    void shouldNotBeValidWithInvalidUrls(final String gatewayUrl) {
+        final IpfsConfiguration config = new IpfsConfiguration(gatewayUrl);
+        assertThat(validator.validate(config))
+                .isNotEmpty()
+                .extracting(ConstraintViolation::getMessage)
+                .containsExactly("must be a valid URL");
+    }
+
+    @Test
+    void shouldBeValid() {
+        final IpfsConfiguration config = new IpfsConfiguration("http://localhost");
+        assertThat(validator.validate(config)).isEmpty();
+    }
+}