[fix][relayer][core] isolate receipt confirmation failures

acb-relayer/r-core/pom.xml

@@ -51,6 +51,11 @@
             <artifactId>wsdl4j</artifactId>
             <version>1.6.3</version>
         </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -128,4 +133,4 @@
         </plugins>
     </build>
 
-</project>
+</project>

acb-relayer/r-core/src/main/java/com/alipay/antchain/bridge/relayer/core/service/confirm/AMConfirmService.java

@@ -108,24 +108,12 @@ public void process(String product, String blockchainId) {
         List<SDPNonceRecordDO> sdpNonceRecordsToSave = new ArrayList<>();
         futureList.forEach(
                 future -> {
-                    ConfirmResult result;
-                    try {
-                        result = future.get();
-                    } catch (InterruptedException | ExecutionException e) {
-                        throw new RuntimeException(
-                                String.format("failed to query cross-chain receipt for ( product: %s, bid: %s )", product, blockchainId),
-                                e
-                        );
+                    ConfirmResult result = resolveConfirmResult(future, product, blockchainId);
+                    if (ObjectUtil.isNull(result)) {
+                        return;
                     }
                     if (result.getReceipt().isConfirmed()) {
-                        SDPMsgCommitResult sdpMsgCommitResult = new SDPMsgCommitResult(
-                                product,
-                                blockchainId,
-                                result.getReceipt().getTxhash(),
-                                result.getReceipt().isSuccessful(),
-                                result.getReceipt().getErrorMsg(),
-                                System.currentTimeMillis()
-                        );
+                        SDPMsgCommitResult sdpMsgCommitResult = buildCommitResult(product, blockchainId, result);
                         if (result.getSdpMsg().getVersion() > 2
                                 && result.getSdpMsg().getSdpMessage().getAtomicFlag().ordinal() < AtomicFlagEnum.ACK_SUCCESS.ordinal()
                                 && result.getSdpMsg().getSdpMessage().getTimeoutMeasure() != TimeoutMeasureEnum.NO_TIMEOUT
@@ -160,6 +148,50 @@ public void process(String product, String blockchainId) {
         );
     }
 
+    static ConfirmResult resolveConfirmResult(
+            Future<ConfirmResult> future,
+            String product,
+            String blockchainId
+    ) {
+        try {
+            return future.get();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            // Interruption is a worker-lifecycle signal rather than a
+            // transaction-local failure. Preserve it and abort this pass.
+            throw new RuntimeException(
+                    String.format("interrupted while querying cross-chain receipt for ( product: %s, bid: %s )", product, blockchainId),
+                    e
+            );
+        } catch (ExecutionException e) {
+            // Receipt availability is per transaction. One temporarily missing
+            // native receipt must not discard confirmed results from the rest
+            // of this batch; the failed row remains TX_PENDING for reconciliation.
+            log.error(
+                    "failed to query one cross-chain receipt for ( product: {}, bid: {} ), keep it pending",
+                    product,
+                    blockchainId,
+                    e.getCause()
+            );
+        }
+        return null;
+    }
+
+    static SDPMsgCommitResult buildCommitResult(String product, String blockchainId, ConfirmResult result) {
+        // The pending row is already known. Updating by its primary key avoids
+        // depending on a plugin's native transaction-hash representation
+        // (notably Dioxide's non-hex hash) for terminal-state persistence.
+        return new SDPMsgCommitResult(
+                result.getSdpMsg().getId(),
+                product,
+                blockchainId,
+                result.getReceipt().getTxhash(),
+                result.getReceipt().isSuccessful(),
+                result.getReceipt().getErrorMsg(),
+                System.currentTimeMillis()
+        );
+    }
+
     public void processTimeout(String product, String blockchainId) {
         List<SDPMsgWrapper> sdpMsgWrappers = crossChainMessageRepository.peekSDPMessagesSent(
                 product,

acb-relayer/r-core/src/test/java/com/alipay/antchain/bridge/relayer/core/service/confirm/AMConfirmServiceTest.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2024 Ant Group
+ *
+ * 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.alipay.antchain.bridge.relayer.core.service.confirm;
+
+import java.util.concurrent.CompletableFuture;
+
+import com.alipay.antchain.bridge.commons.core.base.CrossChainMessageReceipt;
+import com.alipay.antchain.bridge.relayer.commons.model.SDPMsgCommitResult;
+import com.alipay.antchain.bridge.relayer.commons.model.SDPMsgWrapper;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class AMConfirmServiceTest {
+
+    @Test
+    public void failedReceiptShouldRemainPendingWithoutAbortingTheBatch() {
+        CompletableFuture<ConfirmResult> failed = new CompletableFuture<>();
+        failed.completeExceptionally(new RuntimeException("native transaction temporarily unavailable"));
+
+        Assert.assertNull(AMConfirmService.resolveConfirmResult(failed, "dioxide2", "diox04.id"));
+    }
+
+    @Test
+    public void interruptedReceiptShouldAbortAndPreserveInterruptFlag() {
+        CompletableFuture<ConfirmResult> pending = new CompletableFuture<>();
+        Thread.currentThread().interrupt();
+        try {
+            AMConfirmService.resolveConfirmResult(pending, "dioxide2", "diox04.id");
+            Assert.fail("expected the interrupted confirmation pass to abort");
+        } catch (RuntimeException expected) {
+            Assert.assertTrue(Thread.currentThread().isInterrupted());
+            Assert.assertTrue(expected.getCause() instanceof InterruptedException);
+        } finally {
+            Thread.interrupted();
+        }
+    }
+
+    @Test
+    public void buildCommitResultShouldKeepPendingRowIdentityForNativeHash() {
+        SDPMsgWrapper message = new SDPMsgWrapper();
+        message.setId(2700L);
+
+        CrossChainMessageReceipt receipt = new CrossChainMessageReceipt();
+        receipt.setTxhash("yh7jj8ntyz8ervas66rfv04e7zsjxc6x4ftedvdqk2hz4kbatq7g");
+        receipt.setSuccessful(true);
+        receipt.setConfirmed(true);
+        receipt.setErrorMsg("");
+
+        SDPMsgCommitResult result = AMConfirmService.buildCommitResult(
+                "dioxide2",
+                "diox04.id",
+                new ConfirmResult(receipt, message)
+        );
+
+        Assert.assertEquals(Long.valueOf(2700L), result.getSdpMsgId());
+        Assert.assertEquals(receipt.getTxhash(), result.getTxHash());
+        Assert.assertTrue(result.isCommitSuccess());
+    }
+}