Fix monitor persistence and cross-chain receipt finality

acb-relayer/README.md

@@ -101,6 +101,8 @@ docker run -itd --name redis-test -p 6379:6379 redis --requirepass 'YOUR_PWD' --
 
 在开始之前，需要初始化中继的数据库，这里提供一个[DDL](r-bootstrap/src/main/resources/db/ddl.sql)，或者解压之后在路径`config/db/ddl.sql`找到，在MySQL执行即可生成数据库`relayer`。
 
+已有数据库升级时，需要按版本执行[升级脚本](r-bootstrap/src/main/resources/db/upgrade/README.md)，再启动新版中继服务。
+
 #### TLS
 
 这里初始化中继的TLS证书，会在`tls_certs`路径下生成`relayer.crt`和`relayer.key`。

acb-relayer/r-bootstrap/desc_tar.xml

@@ -28,6 +28,7 @@
             <includes>
                 <include>application.yml</include>
                 <include>db${file.separator}ddl.sql</include>
+                <include>db${file.separator}upgrade${file.separator}**</include>
             </includes>
         </fileSet>
         <fileSet>
@@ -38,4 +39,4 @@
             </includes>
         </fileSet>
     </fileSets>
-</assembly>
+</assembly>

acb-relayer/r-bootstrap/src/main/resources/db/ddl.sql

@@ -56,7 +56,7 @@ CREATE TABLE IF NOT EXISTS `anchor_process`
     `gmt_create`         datetime            DEFAULT CURRENT_TIMESTAMP,
     `gmt_modified`       datetime            DEFAULT CURRENT_TIMESTAMP,
     PRIMARY KEY (`id`),
-    UNIQUE KEY `blockchain_product` (`blockchain_product`, `instance`, `task`)
+    UNIQUE KEY `blockchain_product` (`blockchain_product`, `instance`, `task`, `tpbta_lane_key`)
 ) ENGINE = InnoDB
   ROW_FORMAT = DYNAMIC;
 
@@ -545,4 +545,4 @@ CREATE TABLE IF NOT EXISTS `sdp_nonce_record`
     PRIMARY KEY (`id`),
     KEY `idx_message_id_nonce` (`message_id`),
     KEY `idx_channel` (`sender_domain`, `sender_identity`, `receiver_domain`, `receiver_identity`, `nonce`)
-);
+);

acb-relayer/r-bootstrap/src/main/resources/db/upgrade/20260820_anchor_process_lane_index.sql

@@ -0,0 +1,4 @@
+ALTER TABLE `anchor_process`
+    DROP INDEX `blockchain_product`,
+    ADD UNIQUE KEY `blockchain_product`
+        (`blockchain_product`, `instance`, `task`, `tpbta_lane_key`);

acb-relayer/r-bootstrap/src/main/resources/db/upgrade/README.md

@@ -0,0 +1,7 @@
+# 数据库升级
+
+已有数据库不会重复执行 `ddl.sql` 中的表定义。升级中继服务前，请按版本顺序手动执行本目录中的 SQL 脚本，并在执行前备份数据库。
+
+## 2026-08-20
+
+执行 `20260820_anchor_process_lane_index.sql`，使不同 TPBTA lane 的锚定任务能够分别保存进度。

acb-relayer/r-bootstrap/src/test/resources/data/ddl.sql

@@ -51,7 +51,7 @@ CREATE TABLE IF NOT EXISTS `anchor_process`
     `gmt_create`         datetime     DEFAULT CURRENT_TIMESTAMP,
     `gmt_modified`       datetime     DEFAULT CURRENT_TIMESTAMP,
     PRIMARY KEY (`id`),
-    UNIQUE KEY `blockchain_product` (`blockchain_product`, `instance`, `task`)
+    UNIQUE KEY `blockchain_product` (`blockchain_product`, `instance`, `task`, `tpbta_lane_key`)
 );
 
 CREATE TABLE IF NOT EXISTS `domain_cert`
@@ -503,4 +503,4 @@ CREATE TABLE IF NOT EXISTS `sdp_nonce_record`
     `hash_val`          VARCHAR(64) UNIQUE  NOT NULL,
     `gmt_create`        DATETIME DEFAULT CURRENT_TIMESTAMP,
     `gmt_modified`      DATETIME DEFAULT CURRENT_TIMESTAMP
-);
+);

acb-relayer/r-commons/src/main/java/com/alipay/antchain/bridge/relayer/commons/model/BlockchainMeta.java

@@ -168,6 +168,12 @@ public void updateProperties(BlockchainProperties properties) {
         if (StrUtil.isNotEmpty(properties.getSdpMsgContractAddress())) {
             this.properties.setSdpMsgContractAddress(properties.getSdpMsgContractAddress());
         }
+        if (StrUtil.isNotEmpty(properties.getPtcContractAddress())) {
+            this.properties.setPtcContractAddress(properties.getPtcContractAddress());
+        }
+        if (StrUtil.isNotEmpty(properties.getMonitorContractAddress())) {
+            this.properties.setMonitorContractAddress(properties.getMonitorContractAddress());
+        }
         if (ObjectUtil.isNotNull(properties.getAnchorRuntimeStatus())) {
             this.properties.setAnchorRuntimeStatus(properties.getAnchorRuntimeStatus());
         }

acb-relayer/r-core/src/main/java/com/alipay/antchain/bridge/relayer/core/manager/blockchain/BlockchainManager.java

@@ -202,7 +202,7 @@ public void updateBlockchain(String product, String blockchainId, String pluginS
             blockchainMeta.setAlias(alias);
             blockchainMeta.setDesc(desc);
             blockchainMeta.updateProperties(blockchainProperties);
-            if (!updateBlockchainMeta(new BlockchainMeta(product, blockchainId, alias, desc, blockchainProperties))) {
+            if (!updateBlockchainMeta(blockchainMeta)) {
                 throw new RuntimeException(
                         StrUtil.format(
                                 "failed to update meta for blockchain {} - {} into DB",
@@ -561,11 +561,40 @@ public void checkTpBtaReadyOnReceivingChain(SDPMsgWrapper sdpMsgWrapper) {
             return;
         }
 
-        if (blockchainRepository.hasBta(sdpMsgWrapper.getSdpMessage().getTargetDomain())) {
+        // The third-party proof and its TpBTA describe the sending blockchain.
+        // Checking the target domain here skips TpBTA upload whenever only the
+        // source blockchain is trust-enabled, and the subsequent target-chain
+        // submission then fails with "tpbta not found".
+        if (blockchainRepository.hasBta(new CrossChainDomain(sdpMsgWrapper.getSenderBlockchainDomain()))) {
+            AbstractBlockchainClient receivingClient = blockchainClientPool.getClient(
+                    sdpMsgWrapper.getReceiverBlockchainProduct(),
+                    sdpMsgWrapper.getReceiverBlockchainId()
+            );
+            if (ObjectUtil.isNotNull(receivingClient)
+                    && !hasConfiguredPtcContract(receivingClient.queryBBCContext())) {
+                // Some V0 BBC implementations (currently Dioxide) intentionally
+                // verify the relayed AM without an on-chain PTC Hub. Calling the
+                // V1 hasTpBta API for those targets fails before relayAuthMessage
+                // can run. Trust synchronization is only applicable when the
+                // receiving chain actually exposes a configured PTC contract.
+                log.info(
+                        "receiving blockchain {}-{} has no configured ptc contract, skip tpbta upload check",
+                        sdpMsgWrapper.getReceiverBlockchainProduct(),
+                        sdpMsgWrapper.getReceiverBlockchainId()
+                );
+                return;
+            }
             checkTpBtaReadyOnReceivingChain(sdpMsgWrapper, tpProof);
         }
     }
 
+    static boolean hasConfiguredPtcContract(AbstractBBCContext bbcContext) {
+        return ObjectUtil.isNotNull(bbcContext)
+                && ObjectUtil.isNotNull(bbcContext.getPtcContract())
+                && StrUtil.isNotBlank(bbcContext.getPtcContract().getContractAddress())
+                && !"empty".equalsIgnoreCase(bbcContext.getPtcContract().getContractAddress());
+    }
+
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void initTpBta(@NonNull String ptcServiceId, @NonNull IBlockchainTrustAnchor bta) {
@@ -923,38 +952,29 @@ private void deployHeteroBlockchainAMContract(BlockchainMeta blockchainMeta) {
             blockchainClient.getAMClientContract().deployContract();
         }
 
-        if (
-                ObjectUtil.isNull(bbcContext.getSdpContract())
-                        || ContractStatusEnum.INIT == bbcContext.getSdpContract().getStatus()
-        ) {
-            blockchainClient.getSDPMsgClientContract().deployContract();
-        }
+        // setup-bbccontracts is also the explicit reconciliation entry point
+        // for versioned system-contract upgrades. BBC implementations must
+        // keep setup idempotent when an existing deployment is current.
+        blockchainClient.getSDPMsgClientContract().deployContract();
 
         // if monitor is supported, it needs to be deployed before ptc
         boolean isMonitorSupport = true;
-        if (
-                ObjectUtil.isNull(bbcContext.getMonitorContract())
-                        || ContractStatusEnum.INIT == bbcContext.getMonitorContract().getStatus()
-        ) {
-            try {
-                blockchainClient.getMonitorClientContract().deployContract();
-            } catch (BbcInterfaceNotSupportException e) {
-                log.info("setup monitor contract not support for blockchain product: {}", blockchainMeta.getProduct());
-                isMonitorSupport = false;
-            }
+        try {
+            blockchainClient.getMonitorClientContract().deployContract();
+        } catch (BbcInterfaceNotSupportException e) {
+            log.info("setup monitor contract not support for blockchain product: {}", blockchainMeta.getProduct());
+            isMonitorSupport = false;
         }
 
         boolean isPtcSupport = true;
-        if (
-                ObjectUtil.isNull(bbcContext.getPtcContract())
-                        || ContractStatusEnum.INIT == bbcContext.getPtcContract().getStatus()
-        ) {
-            try {
-                blockchainClient.getPtcContract().deployContract();
-            } catch (BbcInterfaceNotSupportException e) {
-                log.info("setup ptc contract not support for blockchain product: {}", blockchainMeta.getProduct());
-                isPtcSupport = false;
-            }
+        try {
+            // setup-bbccontracts is an explicit reconciliation request. The
+            // BBC implementation must be allowed to probe and replace a
+            // legacy PTC Hub even when the persisted context says READY.
+            blockchainClient.getPtcContract().deployContract();
+        } catch (BbcInterfaceNotSupportException e) {
+            log.info("setup ptc contract not support for blockchain product: {}", blockchainMeta.getProduct());
+            isPtcSupport = false;
         }
 
         bbcContext = blockchainClient.queryBBCContext();

acb-relayer/r-core/src/main/java/com/alipay/antchain/bridge/relayer/core/service/anchor/workers/ConsensusStateWorker.java

@@ -16,10 +16,16 @@
 
 package com.alipay.antchain.bridge.relayer.core.service.anchor.workers;
 
+import java.math.BigInteger;
+
 import cn.hutool.core.util.ObjectUtil;
+import com.alipay.antchain.bridge.commons.core.base.CrossChainDomain;
+import com.alipay.antchain.bridge.commons.core.base.ConsensusState;
+import com.alipay.antchain.bridge.commons.core.bta.IBlockchainTrustAnchor;
 import com.alipay.antchain.bridge.commons.core.ptc.ValidatedConsensusState;
 import com.alipay.antchain.bridge.ptc.service.IPTCService;
 import com.alipay.antchain.bridge.relayer.commons.exception.PtcVerifyConsensusStateException;
+import com.alipay.antchain.bridge.relayer.commons.model.BtaDO;
 import com.alipay.antchain.bridge.relayer.commons.model.TpBtaDO;
 import com.alipay.antchain.bridge.relayer.commons.model.ValidatedConsensusStateDO;
 import com.alipay.antchain.bridge.relayer.core.service.anchor.context.AnchorProcessContext;
@@ -55,6 +61,8 @@ public boolean process(AbstractBlock block, TpBtaDO tpBtaDO) {
                 throw new PtcVerifyConsensusStateException("ptc service stub {} is null", tpBtaDO.getPtcServiceId());
             }
 
+            restoreAnchorStateIfNecessary(block, tpBtaDO, ptcService);
+
             ValidatedConsensusStateDO parentVcs = ValidatedConsensusStateDO.builder().build();
             if (!ptcService.getPtcFeatureDescriptor().isStorageEnabled()) {
                 parentVcs = getProcessContext().getBlockchainRepository().getValidatedConsensusState(
@@ -115,4 +123,55 @@ public boolean process(AbstractBlock block, TpBtaDO tpBtaDO) {
 
         return true;
     }
+
+    /**
+     * A storage-enabled PTC keeps the verified consensus-state chain on the PTC node. If that
+     * runtime storage is restored from an incomplete snapshot while the relayer database keeps
+     * its anchor progress, the first post-BTA block can no longer be verified because its parent
+     * (the BTA anchor state) is missing.
+     *
+     * Re-submitting the BTA anchor is safe and idempotent on committee PTC nodes. Limit the repair
+     * strictly to {@code initHeight + 1}; a missing parent at any later height is a real continuity
+     * error and must not be hidden by rewinding or fabricating state.
+     */
+    private void restoreAnchorStateIfNecessary(AbstractBlock block, TpBtaDO tpBtaDO, IPTCService ptcService) {
+        if (!ptcService.getPtcFeatureDescriptor().isStorageEnabled()) {
+            return;
+        }
+
+        BtaDO btaDO = getProcessContext().getBlockchainRepository().getBta(new CrossChainDomain(block.getDomain()));
+        if (ObjectUtil.isNull(btaDO) || ObjectUtil.isNull(btaDO.getBta())) {
+            throw new PtcVerifyConsensusStateException("bta is not found for domain {}", block.getDomain());
+        }
+
+        IBlockchainTrustAnchor bta = btaDO.getBta();
+        if (bta.getSubjectVersion() != tpBtaDO.getBtaSubjectVersion()) {
+            throw new PtcVerifyConsensusStateException(
+                    "bta subject version {} does not match tpbta subject version {} for domain {}",
+                    bta.getSubjectVersion(), tpBtaDO.getBtaSubjectVersion(), block.getDomain()
+            );
+        }
+        if (!BigInteger.valueOf(block.getHeight()).equals(bta.getInitHeight().add(BigInteger.ONE))) {
+            return;
+        }
+
+        ConsensusState anchorState = getProcessContext().getBlockchainClient().getConsensusState(bta.getInitHeight());
+        if (ObjectUtil.isNull(anchorState)) {
+            throw new PtcVerifyConsensusStateException(
+                    "failed to query BTA anchor consensus state at height {} for domain {}",
+                    bta.getInitHeight(), block.getDomain()
+            );
+        }
+        ValidatedConsensusState anchorVcs = ptcService.commitAnchorState(bta, tpBtaDO.getTpbta(), anchorState);
+        if (ObjectUtil.isNull(anchorVcs)) {
+            throw new PtcVerifyConsensusStateException(
+                    "ptc {} returned null while restoring BTA anchor state at height {} for domain {}",
+                    tpBtaDO.getPtcServiceId(), bta.getInitHeight(), block.getDomain()
+            );
+        }
+        log.info(
+                "ensured BTA anchor consensus state {}-{} for domain {} on storage-enabled ptc {}",
+                anchorState.getHeight(), anchorState.getHashHex(), block.getDomain(), tpBtaDO.getPtcServiceId()
+        );
+    }
 }

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

@@ -112,24 +112,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
@@ -171,6 +159,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();
+            log.error(
+                    "interrupted while querying cross-chain receipt for ( product: {}, bid: {} )",
+                    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/main/java/com/alipay/antchain/bridge/relayer/core/types/pluginserver/GRpcBBCServiceClient.java

@@ -25,6 +25,8 @@
 
 public class GRpcBBCServiceClient implements IBBCServiceClient {
 
+    static final long RECEIPT_QUERY_DEADLINE_SECONDS = 30L;
+
     private String psId;
 
     private final String product;
@@ -108,7 +110,7 @@ public AbstractBBCContext getContext() {
 
     @Override
     public CrossChainMessageReceipt readCrossChainMessageReceipt(String txhash) {
-        Response response = this.blockingStub.bbcCall(
+        Response response = withReceiptQueryDeadline(this.blockingStub).bbcCall(
                 CallBBCRequest.newBuilder()
                         .setProduct(this.getProduct())
                         .setDomain(this.getDomain())
@@ -127,6 +129,12 @@ public CrossChainMessageReceipt readCrossChainMessageReceipt(String txhash) {
         );
     }
 
+    static CrossChainServiceGrpc.CrossChainServiceBlockingStub withReceiptQueryDeadline(
+            CrossChainServiceGrpc.CrossChainServiceBlockingStub stub
+    ) {
+        return stub.withDeadlineAfter(RECEIPT_QUERY_DEADLINE_SECONDS, TimeUnit.SECONDS);
+    }
+
     @Override
     public List<CrossChainMessage> readCrossChainMessagesByHeight(long height) {
         Response response = this.blockingStub.bbcCall(