fix(dioxide): coordinate concurrent ISNs across plugins and Python callers

acb-pluginserver/ps-server/src/main/java/com/alipay/antchain/bridge/pluginserver/server/CrossChainServiceImpl.java

@@ -446,7 +446,7 @@ private Response handleSetPtcContract(IBBCService bbcService, SetPtcContractRequ
 
     private Response handleRelayAuthMessage(IBBCService bbcService, RelayAuthMessageRequest request, String product, String domain) {
         try {
-            CrossChainMessageReceipt ret = bbcService.relayAuthMessage(request.getRawMessage().toByteArray());
+            CrossChainMessageReceipt ret = bbcService.relayAuthMessage(request.getRawMessage().toByteArray(), request.getSubmissionId());
             return ResponseBuilder.buildBBCSuccessResp(CallBBCResponse.newBuilder()
                     .setRelayAuthMessageResponse(RelayAuthMessageResponse.newBuilder()
                             .setReceipt(

acb-pluginserver/ps-service/src/main/proto/pluginserver.proto

@@ -160,6 +160,7 @@ message SetPtcContractRequest {
 
 message RelayAuthMessageRequest {
   bytes rawMessage = 1;
+  string submissionId = 2;
 }
 
 message SetAmContractRequest {
@@ -429,4 +430,4 @@ message ReliableRetryResponse {
 
 message RelayMonitorOrderResponse {
   CrossChainMessageReceipt receipt = 1;
-}
+}

acb-relayer/r-core/src/main/java/com/alipay/antchain/bridge/relayer/core/manager/bbc/AMClientContractHeteroBlockchainImpl.java

@@ -24,7 +24,14 @@ public AMClientContractHeteroBlockchainImpl(IBBCServiceClient bbcServiceClient)
     @Override
     public SendResponseResult recvPkgFromRelayer(AuthMsgPackage pkg) {
         try {
-            CrossChainMessageReceipt receipt = bbcServiceClient.relayAuthMessage(pkg.extractProofs());
+            String submissionId = "";
+            if (pkg.getSdpMsgWrapper() != null && pkg.getSdpMsgWrapper().getAuthMsgWrapper() != null) {
+                String ucpId = pkg.getSdpMsgWrapper().getAuthMsgWrapper().getUcpId();
+                if (ucpId != null && !ucpId.isEmpty()) {
+                    submissionId = stableSubmissionId(ucpId, bbcServiceClient.getDomain());
+                }
+            }
+            CrossChainMessageReceipt receipt = bbcServiceClient.relayAuthMessage(pkg.extractProofs(), submissionId);
             if (ObjectUtil.isNull(receipt)) {
                 return new SendResponseResult(
                         "",
@@ -77,6 +84,10 @@ public void setProtocol(String protocolContract, String protocolType) {
         this.bbcServiceClient.setProtocol(protocolContract, protocolType);
     }
 
+    static String stableSubmissionId(String ucpId, String domain) {
+        return cn.hutool.crypto.digest.DigestUtil.sha256Hex("relay-am|" + ucpId + "|" + domain);
+    }
+
     @Override
     public void setPtcContract(String ptcContractAddress) {
         this.bbcServiceClient.setPtcContract(ptcContractAddress);

acb-relayer/r-core/src/main/java/com/alipay/antchain/bridge/relayer/core/types/pluginserver/GRpcBBCServiceClient.java

@@ -350,13 +350,19 @@ public void setPtcContract(String ptcContractAddress) {
 
     @Override
     public CrossChainMessageReceipt relayAuthMessage(byte[] rawMessage) {
+        return relayAuthMessage(rawMessage, "");
+    }
+
+    @Override
+    public CrossChainMessageReceipt relayAuthMessage(byte[] rawMessage, String submissionId) {
         Response response = this.blockingStub.withDeadlineAfter(30, TimeUnit.SECONDS).bbcCall(
                 CallBBCRequest.newBuilder()
                         .setProduct(this.getProduct())
                         .setDomain(this.getDomain())
                         .setRelayAuthMessageReq(
                                 RelayAuthMessageRequest.newBuilder()
                                         .setRawMessage(ByteString.copyFrom(rawMessage))
+                                        .setSubmissionId(submissionId == null ? "" : submissionId)
                         ).build()
         );
         if (response.getCode() != 0) {

acb-relayer/r-core/src/main/proto/pluginserver.proto

@@ -160,6 +160,7 @@ message SetPtcContractRequest {
 
 message RelayAuthMessageRequest {
   bytes rawMessage = 1;
+  string submissionId = 2;
 }
 
 message SetAmContractRequest {
@@ -429,4 +430,4 @@ message ReliableRetryResponse {
 
 message RelayMonitorOrderResponse {
   CrossChainMessageReceipt receipt = 1;
-}
+}

acb-relayer/r-core/src/test/java/com/alipay/antchain/bridge/relayer/core/manager/bbc/SubmissionIdentityTest.java

@@ -0,0 +1,14 @@
+package com.alipay.antchain.bridge.relayer.core.manager.bbc;
+
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+public class SubmissionIdentityTest {
+    @Test public void identityUsesUcpAndTargetNotPayloadOrZeroMessageId() {
+        String first = AMClientContractHeteroBlockchainImpl.stableSubmissionId("ucp-a", "diox04");
+        assertEquals(first, AMClientContractHeteroBlockchainImpl.stableSubmissionId("ucp-a", "diox04"));
+        assertNotEquals(first, AMClientContractHeteroBlockchainImpl.stableSubmissionId("ucp-b", "diox04"));
+        assertNotEquals(first, AMClientContractHeteroBlockchainImpl.stableSubmissionId("ucp-a", "diox11"));
+        assertEquals(64, first.length());
+    }
+}

acb-sdk/antchain-bridge-plugin-lib/pom.xml

@@ -15,6 +15,12 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>junit</groupId><artifactId>junit</artifactId><version>4.13.2</version><scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.mysql</groupId><artifactId>mysql-connector-j</artifactId><version>8.4.0</version><scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.pf4j</groupId>
             <artifactId>pf4j</artifactId>
@@ -61,4 +67,4 @@
             <url>https://maven.pkg.github.com/AntChainOpenLab/AntChainBridgePluginSDK</url>
         </repository>
     </distributionManagement>
-</project>
+</project>

acb-sdk/antchain-bridge-plugin-lib/src/main/java/com/alipay/antchain/bridge/plugins/lib/transactions/JdbcTransactionCoordinator.java

@@ -0,0 +1,229 @@
+package com.alipay.antchain.bridge.plugins.lib.transactions;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.io.InputStream;
+import java.security.MessageDigest;
+import java.sql.*;
+import java.util.Properties;
+import java.util.concurrent.Callable;
+
+/**
+ * Durable account-scoped allocation, shared across plugin class loaders and hosts.
+ * Schema is installed explicitly, never by a transaction-serving process.
+ * Signed bytes are committed BEFORE broadcast; uncertain submissions never allocate again.
+ */
+public final class JdbcTransactionCoordinator {
+    public static final long MAX_ISN = 0xffff_ffffL;
+
+    public interface Transport {
+        String checkpoint() throws Exception;
+        long currentIsn(String account) throws Exception;
+        byte[] composeAndSign(long isn) throws Exception;
+        String broadcast(byte[] signed) throws Exception;
+    }
+
+    public interface Connections { Connection open() throws SQLException; }
+
+    private final Connections connections;
+    private final String network;
+    private final String checkpoint;
+
+    public JdbcTransactionCoordinator(Connections connections, String network, String checkpoint) {
+        require(network, 96, "network");
+        require(checkpoint, 128, "checkpoint");
+        this.connections = connections;
+        this.network = network;
+        this.checkpoint = checkpoint;
+    }
+
+    public static Properties readConfig(String filename) throws Exception {
+        if (filename == null || filename.trim().isEmpty()) {
+            throw new IllegalStateException("DIOXIDE_TX_COORDINATOR_CONFIG must be configured; unsafe allocation is disabled");
+        }
+        Properties p = new Properties();
+        try (InputStream in = Files.newInputStream(Paths.get(filename))) { p.load(in); }
+        return p;
+    }
+
+    public static JdbcTransactionCoordinator fromProperties(Properties p) throws Exception {
+        return fromProperties(p, Thread.currentThread().getContextClassLoader());
+    }
+
+    public static JdbcTransactionCoordinator fromProperties(Properties p, ClassLoader pluginLoader) throws Exception {
+        final String url = required(p, "jdbcUrl");
+        final Properties credentials = new Properties();
+        credentials.setProperty("user", required(p, "user"));
+        credentials.setProperty("password", new String(Files.readAllBytes(Paths.get(
+                required(p, "passwordFile"))), StandardCharsets.UTF_8).trim());
+        // DriverManager's caller class-loader filtering is unsuitable for PF4J plugin drivers.
+        final Driver driver = (Driver) Class.forName("com.mysql.cj.jdbc.Driver", true, pluginLoader).newInstance();
+        return new JdbcTransactionCoordinator(() -> {
+            Connection c = driver.connect(url, credentials);
+            if (c == null) { throw new SQLException("unsupported transaction coordinator JDBC URL"); }
+            return c;
+        }, required(p, "networkId"), required(p, "checkpointHash"));
+    }
+
+    public static String required(Properties p, String name) {
+        String value = p.getProperty(name);
+        if (value == null || value.trim().isEmpty()) { throw new IllegalArgumentException("missing coordinator setting: " + name); }
+        return value.trim();
+    }
+
+    public static String sha256(byte[] input) {
+        try {
+            byte[] bytes = MessageDigest.getInstance("SHA-256").digest(input);
+            StringBuilder out = new StringBuilder(64);
+            for (byte b : bytes) { out.append(String.format("%02x", b & 255)); }
+            return out.toString();
+        } catch (Exception e) { throw new IllegalStateException(e); }
+    }
+
+    public static long nextIsn(long node, long persisted) {
+        if (node < 0 || node > MAX_ISN || persisted < 0 || persisted > MAX_ISN) {
+            throw new IllegalStateException("ISN exhausted or invalid; refusing wraparound");
+        }
+        return Math.max(node, persisted);
+    }
+
+    private static void require(String value, int limit, String field) {
+        if (value == null || value.isEmpty() || value.length() > limit || !value.matches("[\\x21-\\x7e]+")) {
+            throw new IllegalArgumentException("invalid coordinator " + field);
+        }
+    }
+
+    private String lockName(String kind, String account) {
+        return sha256((kind + "|" + network + "|" + account).getBytes(StandardCharsets.UTF_8));
+    }
+
+    public static String normalizeAccount(String account) {
+        if (account != null && account.toLowerCase(java.util.Locale.ROOT).endsWith(":ed25519")) {
+            return account.substring(0, account.length() - 8).toLowerCase(java.util.Locale.ROOT);
+        }
+        return account;
+    }
+
+    private static void acquire(Connection c, String name) throws SQLException {
+        try (PreparedStatement s = c.prepareStatement("SELECT GET_LOCK(?, 20)")) {
+            s.setString(1, name);
+            try (ResultSet r = s.executeQuery()) {
+                if (!r.next() || r.getInt(1) != 1 || r.wasNull()) { throw new SQLException("coordinator lock timeout"); }
+            }
+        }
+    }
+
+    private static void release(Connection c, String name) {
+        try (PreparedStatement s = c.prepareStatement("SELECT RELEASE_LOCK(?)")) {
+            s.setString(1, name); s.executeQuery().close();
+        } catch (SQLException ignored) { /* Closing the connection also releases its named locks. */ }
+    }
+
+    /** Used only around the legacy shared SDP query mailbox, never normal confirmation polling. */
+    public <T> T withQueryLock(String contractAndAccount, Callable<T> query) throws Exception {
+        String name = lockName("query", contractAndAccount);
+        try (Connection c = connections.open()) {
+            acquire(c, name);
+            try { return query.call(); } finally { release(c, name); }
+        }
+    }
+
+    public String submit(String operationId, String account, byte[] payload, Transport transport) throws Exception {
+        account = normalizeAccount(account);
+        require(operationId, 191, "operationId");
+        require(account, 160, "account");
+        String name = lockName("submission", account);
+        String fingerprint = sha256(payload);
+        try (Connection c = connections.open()) {
+            acquire(c, name);
+            try {
+                if (!checkpoint.equals(transport.checkpoint())) {
+                    throw new IllegalStateException("Dioxide network checkpoint changed; submission disabled");
+                }
+                c.setAutoCommit(false);
+                byte[] signed;
+                try {
+                    try (PreparedStatement s = c.prepareStatement(
+                            "INSERT INTO bridge_tx_account(network_id,account,checkpoint_hash,next_isn) VALUES(?,?,?,0) " +
+                            "ON DUPLICATE KEY UPDATE account=VALUES(account)")) {
+                        s.setString(1, network); s.setString(2, account); s.setString(3, checkpoint); s.executeUpdate();
+                    }
+                    long persisted;
+                    long observed;
+                    try (PreparedStatement s = c.prepareStatement(
+                            "SELECT checkpoint_hash,next_isn,observed_isn FROM bridge_tx_account WHERE network_id=? AND account=? FOR UPDATE")) {
+                        s.setString(1, network); s.setString(2, account);
+                        try (ResultSet r = s.executeQuery()) {
+                            if (!r.next() || !checkpoint.equals(r.getString(1))) { throw new IllegalStateException("coordinator network mismatch"); }
+                            persisted = r.getLong(2);
+                            observed = r.getLong(3);
+                        }
+                    }
+                    try (PreparedStatement s = c.prepareStatement(
+                            "SELECT account,payload_hash,signed_tx,tx_hash FROM bridge_tx_submission WHERE network_id=? AND operation_id=? FOR UPDATE")) {
+                        s.setString(1, network); s.setString(2, operationId);
+                        try (ResultSet r = s.executeQuery()) {
+                            if (r.next()) {
+                                if (!account.equals(r.getString(1)) || !fingerprint.equals(r.getString(2))) {
+                                    throw new IllegalStateException("submission identity reused with different account or payload");
+                                }
+                                signed = r.getBytes(3);
+                                String hash = r.getString(4);
+                                c.commit(); c.setAutoCommit(true);
+                                if (hash != null && !hash.isEmpty()) { return hash; }
+                                return broadcast(c, operationId, signed, transport);
+                            }
+                        }
+                    }
+                    long nodeIsn = transport.currentIsn(account);
+                    if (nodeIsn < observed) {
+                        throw new IllegalStateException("node ISN regressed; reconcile network state before new submission");
+                    }
+                    long isn = nextIsn(nodeIsn, persisted);
+                    signed = transport.composeAndSign(isn);
+                    if (signed == null || signed.length == 0) { throw new IllegalStateException("empty signed transaction"); }
+                    try (PreparedStatement s = c.prepareStatement(
+                            "INSERT INTO bridge_tx_submission(network_id,operation_id,account,isn,payload_hash,signed_tx,state) VALUES(?,?,?,?,?,?,'SIGNED')")) {
+                        s.setString(1, network); s.setString(2, operationId); s.setString(3, account);
+                        s.setLong(4, isn); s.setString(5, fingerprint); s.setBytes(6, signed); s.executeUpdate();
+                    }
+                    try (PreparedStatement s = c.prepareStatement(
+                            "UPDATE bridge_tx_account SET next_isn=?,observed_isn=? WHERE network_id=? AND account=?")) {
+                        s.setLong(1, isn + 1); s.setLong(2, nodeIsn); s.setString(3, network); s.setString(4, account); s.executeUpdate();
+                    }
+                    c.commit(); // Never move this below broadcast.
+                } catch (Exception e) {
+                    try { if (!c.getAutoCommit()) { c.rollback(); } } catch (SQLException rollback) { e.addSuppressed(rollback); }
+                    throw e;
+                } finally { c.setAutoCommit(true); }
+                return broadcast(c, operationId, signed, transport);
+            } finally { release(c, name); }
+        }
+    }
+
+    private String broadcast(Connection c, String operationId, byte[] signed, Transport transport) throws Exception {
+        try {
+            String hash = transport.broadcast(signed);
+            if (hash == null || hash.trim().isEmpty()) { throw new IllegalStateException("broadcast returned no hash"); }
+            try (PreparedStatement s = c.prepareStatement(
+                    "UPDATE bridge_tx_submission SET tx_hash=?,state='BROADCAST',last_error=NULL WHERE network_id=? AND operation_id=?")) {
+                s.setString(1, hash); s.setString(2, network); s.setString(3, operationId); s.executeUpdate();
+            }
+            return hash;
+        } catch (Exception e) {
+            try (PreparedStatement s = c.prepareStatement(
+                    "UPDATE bridge_tx_submission SET state='UNKNOWN',last_error=? WHERE network_id=? AND operation_id=? AND tx_hash IS NULL")) {
+                s.setString(1, e.getClass().getSimpleName()); s.setString(2, network); s.setString(3, operationId); s.executeUpdate();
+            } catch (SQLException ignored) { /* SIGNED is also recoverable using the exact stored bytes. */ }
+            throw e;
+        }
+    }
+
+    public void recordOutcome(String hash, boolean success) throws SQLException {
+        try (Connection c = connections.open(); PreparedStatement s = c.prepareStatement(
+                "UPDATE bridge_tx_submission SET state=? WHERE network_id=? AND tx_hash=?")) {
+            s.setString(1, success ? "FINALIZED" : "FAILED"); s.setString(2, network); s.setString(3, hash); s.executeUpdate();
+        }
+    }
+}

acb-sdk/antchain-bridge-plugin-lib/src/main/resources/db/dioxide_tx_coordinator.sql

@@ -0,0 +1,27 @@
+-- Install explicitly in the shared operational database. Do not reset after rollback.
+CREATE TABLE IF NOT EXISTS bridge_tx_account (
+  network_id VARCHAR(96) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
+  account VARCHAR(160) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
+  checkpoint_hash VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
+  next_isn BIGINT NOT NULL,
+  observed_isn BIGINT NOT NULL DEFAULT 0,
+  updated_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
+  PRIMARY KEY (network_id, account)
+) ENGINE=InnoDB;
+CREATE TABLE IF NOT EXISTS bridge_tx_submission (
+  network_id VARCHAR(96) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
+  operation_id VARCHAR(191) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
+  account VARCHAR(160) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
+  isn BIGINT NOT NULL,
+  payload_hash CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL,
+  signed_tx MEDIUMBLOB NOT NULL,
+  tx_hash VARCHAR(128) CHARACTER SET ascii COLLATE ascii_bin NULL,
+  state VARCHAR(16) NOT NULL,
+  last_error VARCHAR(128) NULL,
+  created_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
+  updated_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
+  PRIMARY KEY (network_id, operation_id),
+  UNIQUE KEY uq_bridge_tx_isn (network_id, account, isn),
+  KEY ix_bridge_tx_hash (network_id, tx_hash),
+  KEY ix_bridge_tx_pending (network_id, state)
+) ENGINE=InnoDB;