fix(统一支付): 回调 SSRF 等价字面量/重试快照/并发状态

java/src/main/java/com/payment/platform/module/payment/unified/service/UnifiedCallbackService.java

@@ -28,6 +28,8 @@ import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.client.RestTemplate;
 
+import java.math.BigInteger;
+import java.net.InetAddress;
 import java.net.URI;
 import java.time.OffsetDateTime;
 import java.time.ZoneId;
@@ -36,6 +38,7 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.Executor;
+import java.util.regex.Pattern;
 
 /**
  * 把钱事件（订单/退款）回调给下游商户（spec §7.6）。
@@ -54,8 +57,11 @@ import java.util.concurrent.Executor;
  *   <li><b>{@code content.upstream} 是白名单投影</b>（R35）：上游通知混有我们上送的
  *       {@code channelExtra}（付款码 / openid / buyerUserId）。整体塞原始 JsonNode 等于把
  *       付款凭据转发给下游，命中本模块的硬约束。</li>
- *   <li><b>SSRF 最小防线</b>：只允许 http/https，拒绝本机/环回/云元数据地址；<b>不解析</b>
- *       私网 IP —— 企业内网回调是合法用法。30x 由专用 {@code RestTemplate} 负责不跟随。</li>
+ *   <li><b>SSRF 最小防线</b>：只允许 http/https，拒绝本机/环回/any-local/link-local（含云
+ *       元数据）地址；host 先规范化、数字形态按字面量解析（`localhost.`、
+ *       `[::ffff:127.0.0.1]`、十进制 IPv4 都不能绕过），解析失败 fail-closed。
+ *       <b>不解析</b>私网域名/IP —— 企业内网回调是合法用法。30x 由专用
+ *       {@code RestTemplate} 负责不跟随。</li>
  * </ol>
  *
  * <p>日志约束：url 与异常 message 先过 {@link UnifiedLogSanitizer#flatten}（CWE-117）；
@@ -79,6 +85,9 @@ public class UnifiedCallbackService {
     /** 每轮重试扫描上限。 */
     private static final String RETRY_LIMIT = "LIMIT 50";
 
+    /** IPv6 字面量的字符集（含内嵌 IPv4 的点）；含 {@code %} 的 zone id 一律 fail-closed。 */
+    private static final Pattern IPV6_LITERAL = Pattern.compile("[0-9a-f:.]+");
+
     /**
      * {@code content.upstream} 的白名单（R35，逐字取 spec §7.5）。只拷贝存在的字段；
      * 绝不整体转发原始节点 —— 它可能带 {@code channelExtra} / {@code openid} / {@code sign}。
@@ -118,22 +127,33 @@ public class UnifiedCallbackService {
     /**
      * 订单事件回调。事件矩阵由调用方保证（只在 trade_status 2/3/4/5/6 调用）。
      *
-     * @param upstream 上游通知/查单的 data 节点；重试时传 {@code null}
+     * @param upstream 上游通知/查单的 data 节点；重试时由 {@code upstream_data_json} 还原，
+     *                 该列为空/不可解析则传 {@code null}
      */
     public void notifyOrder(UnifiedOrderEntity order, JsonNode upstream) {
         if (order == null) {
             return;
         }
         String notifyId = orderNotifyId(order.getOrderNo(), order.getTradeStatus());
-        String url = resolveOrderCallbackUrl(order);
-        if (!isAllowedCallbackUrl(url)) {
-            log.warn("[统一支付回调] 订单无可用的回调地址，记 FAILED 待重试: order_no={}, url={}",
-                    order.getOrderNo(), UnifiedLogSanitizer.flatten(url));
+        String url;
+        String content;
+        try {
+            url = resolveOrderCallbackUrl(order);
+            if (!isAllowedCallbackUrl(url)) {
+                log.warn("[统一支付回调] 订单无可用的回调地址，记 FAILED 待重试: order_no={}, url={}",
+                        order.getOrderNo(), UnifiedLogSanitizer.flatten(url));
+                writeOrderNotifyStatus(order, UnifiedNotifyStatus.FAILED, true);
+                return;
+            }
+            content = buildOrderContent(order, upstream);
+        } catch (Exception e) {
+            // 取址/快照构建（含 sumOccupiedAmount 的 DB 调用）抛异常时方法边界不能逃逸：
+            // 行若留在 PENDING，重试扫描（只选 FAILED）永远捞不到它，事件永久丢失。
+            log.warn("[统一支付回调] 准备订单回调异常，记 FAILED 待重试: order_no={}, error={}",
+                    order.getOrderNo(), UnifiedLogSanitizer.flatten(e.getMessage()));
             writeOrderNotifyStatus(order, UnifiedNotifyStatus.FAILED, true);
             return;
         }
-
-        String content = buildOrderContent(order, upstream);
         if (content == null) {
             writeOrderNotifyStatus(order, UnifiedNotifyStatus.FAILED, true);
             return;
@@ -152,22 +172,31 @@ public class UnifiedCallbackService {
      * 退款事件回调。事件矩阵由调用方保证（只在 state 2/3/4 调用）。
      *
      * @param order    退款所属订单（退款事件里要带订单当前快照）；重试时由本类按 order_no 查出
-     * @param upstream 上游通知/查单的 data 节点；重试时传 {@code null}
+     * @param upstream 上游通知/查单的 data 节点；重试时由 {@code upstream_data_json} 还原，
+     *                 该列为空/不可解析则传 {@code null}
      */
     public void notifyRefund(UnifiedRefundEntity refund, UnifiedOrderEntity order, JsonNode upstream) {
         if (refund == null) {
             return;
         }
         String notifyId = refundNotifyId(refund.getRefundNo(), refund.getState());
-        String url = resolveRefundCallbackUrl(refund, order);
-        if (!isAllowedCallbackUrl(url)) {
-            log.warn("[统一支付回调] 退款无可用的回调地址，记 FAILED 待重试: refund_no={}, url={}",
-                    refund.getRefundNo(), UnifiedLogSanitizer.flatten(url));
+        String url;
+        String content;
+        try {
+            url = resolveRefundCallbackUrl(refund, order);
+            if (!isAllowedCallbackUrl(url)) {
+                log.warn("[统一支付回调] 退款无可用的回调地址，记 FAILED 待重试: refund_no={}, url={}",
+                        refund.getRefundNo(), UnifiedLogSanitizer.flatten(url));
+                writeRefundNotifyStatus(refund, UnifiedNotifyStatus.FAILED, true);
+                return;
+            }
+            content = buildRefundContent(refund, order, upstream);
+        } catch (Exception e) {
+            log.warn("[统一支付回调] 准备退款回调异常，记 FAILED 待重试: refund_no={}, error={}",
+                    refund.getRefundNo(), UnifiedLogSanitizer.flatten(e.getMessage()));
             writeRefundNotifyStatus(refund, UnifiedNotifyStatus.FAILED, true);
             return;
         }
-
-        String content = buildRefundContent(refund, order, upstream);
         if (content == null) {
             writeRefundNotifyStatus(refund, UnifiedNotifyStatus.FAILED, true);
             return;
@@ -211,7 +240,7 @@ public class UnifiedCallbackService {
         }
         for (UnifiedOrderEntity order : orders) {
             try {
-                notifyOrder(order, null);
+                notifyOrder(order, parseUpstreamSnapshot(order.getUpstreamDataJson()));
             } catch (Exception e) {
                 // 单行失败不能拖垮整批
                 log.error("[统一支付回调] 重试订单异常: order_no={}, error={}",
@@ -246,7 +275,7 @@ public class UnifiedCallbackService {
                             refund.getRefundNo());
                     continue;
                 }
-                notifyRefund(refund, order, null);
+                notifyRefund(refund, order, parseUpstreamSnapshot(refund.getUpstreamDataJson()));
             } catch (Exception e) {
                 log.error("[统一支付回调] 重试退款异常: refund_no={}, error={}",
                         refund.getRefundNo(), UnifiedLogSanitizer.flatten(e.getMessage()));
@@ -379,6 +408,26 @@ public class UnifiedCallbackService {
         return amount - occupied;
     }
 
+    /**
+     * 重试时把行上持久化的 {@code upstream_data_json} 还原成节点。{@code null}/空串/非法 JSON
+     * 一律当作「没有快照」（{@code content.upstream} 为 {@code {}}），重试本身不因此失败。
+     *
+     * <p>解析失败的异常消息**不打印**：Jackson 的报错会带报文片段，而快照里可能有
+     * {@code channelExtra} / {@code openid}。
+     */
+    private static JsonNode parseUpstreamSnapshot(String json) {
+        if (StrUtil.isBlank(json)) {
+            return null;
+        }
+        try {
+            JsonNode node = CONTENT_MAPPER.readTree(json);
+            return node != null && node.isObject() ? node : null;
+        } catch (Exception e) {
+            log.warn("[统一支付回调] 上游快照 JSON 不可解析，本例 upstream 置空");
+            return null;
+        }
+    }
+
     /**
      * {@code content.upstream} 的白名单投影（R35）：只拷贝存在的字段，绝不整体转发原始节点。
      * 原始上游通知混有我们上送的 {@code channelExtra}（付款码 / openid / buyerUserId）。
@@ -462,12 +511,17 @@ public class UnifiedCallbackService {
      *    会被乐观锁静默挡掉（0 行）；② 计数必须落在 SQL 侧自增，并发重试下
      *    read-modify-write 会丢计数、上限形同虚设。代价是 {@code updated_time} 要手写
      *    （{@code update(null, wrapper)} 不经自动填充）。
+     *
+     * <p><b>WHERE 必须带事件判别键 {@code trade_status}</b>：同一订单行会产生多个 ORDER
+     * 事件（2、2→5）。旧事件的推送如果晚拿到 2xx，只按 {@code order_no} 回写会把新事件的
+     * FAILED 覆盖成 SUCCESS，新事件就永远不再被重试扫描捞起。
      */
     private void writeOrderNotifyStatus(UnifiedOrderEntity order, String status, boolean countAttempt) {
         try {
             OffsetDateTime now = OffsetDateTime.now();
             LambdaUpdateWrapper<UnifiedOrderEntity> update = new LambdaUpdateWrapper<UnifiedOrderEntity>()
                     .eq(UnifiedOrderEntity::getOrderNo, order.getOrderNo())
+                    .eq(UnifiedOrderEntity::getTradeStatus, order.getTradeStatus())
                     .set(UnifiedOrderEntity::getNotifyStatus, status)
                     .set(UnifiedOrderEntity::getLastNotifyTime, now)
                     .set(UnifiedOrderEntity::getUpdatedTime, now);
@@ -483,11 +537,13 @@ public class UnifiedCallbackService {
         }
     }
 
+    /** 退款侧同理：WHERE 带事件判别键 {@code state}。 */
     private void writeRefundNotifyStatus(UnifiedRefundEntity refund, String status, boolean countAttempt) {
         try {
             OffsetDateTime now = OffsetDateTime.now();
             LambdaUpdateWrapper<UnifiedRefundEntity> update = new LambdaUpdateWrapper<UnifiedRefundEntity>()
                     .eq(UnifiedRefundEntity::getRefundNo, refund.getRefundNo())
+                    .eq(UnifiedRefundEntity::getState, refund.getState())
                     .set(UnifiedRefundEntity::getNotifyStatus, status)
                     .set(UnifiedRefundEntity::getLastNotifyTime, now)
                     .set(UnifiedRefundEntity::getUpdatedTime, now);
@@ -507,8 +563,22 @@ public class UnifiedCallbackService {
 
     /**
      * 下游地址的最小 SSRF 防线（spec §7.6）：只允许 http/https 且 host 非空；拒绝
-     * localhost / 127.0.0.0/8 / ::1 / 0.0.0.0 / 云元数据地址。<b>不解析私网 IP</b> ——
-     * 企业内网回调是合法用法，全禁会误伤。DNS 解析也不做（见 spec §11 残余风险）。
+     * localhost / 环回 / any-local / link-local（含云元数据 169.254.169.254）。
+     * <b>不解析私网 IP</b> —— 企业内网回调是合法用法，全禁会误伤。DNS 解析也不做
+     * （见 spec §11 残余风险）。
+     *
+     * <p>光做字符串相等/前缀比较会被等价字面量绕过（`localhost.`、`[::ffff:127.0.0.1]`、
+     * 十进制/前导零 IPv4）。这里的顺序是：
+     *
+     * <ol>
+     *   <li>host 先规范化：trim、小写、去尾部所有点、去 IPv6 方括号；</li>
+     *   <li>非数字 host：只额外拒绝 {@code localhost} 与 {@code *.localhost}，其余放行；</li>
+     *   <li>数字形态 host（纯数字 / {@code 0x} 开头 / 只含数字与点 / 含冒号的 IPv6）：
+     *       按**字面量**解析后再判地址类别，解析失败一律拒绝（fail-closed）。</li>
+     * </ol>
+     *
+     * <p>这里只会对已在字符集上验证过的 IPv6 字面量调用 {@code InetAddress.getByName}；
+     * 域名分支完全不触发 DNS。
      */
     static boolean isAllowedCallbackUrl(String url) {
         if (StrUtil.isBlank(url)) {
@@ -528,28 +598,151 @@ public class UnifiedCallbackService {
         if (!"http".equals(scheme) && !"https".equals(scheme)) {
             return false;
         }
-        String host = uri.getHost();
+        String host = normalizeHost(uri.getHost());
         if (StrUtil.isBlank(host)) {
             return false;
         }
-        // 字面 IPv6 的 getHost() 带方括号：[::1]
-        if (host.startsWith("[") && host.endsWith("]")) {
+        if ("localhost".equals(host) || host.endsWith(".localhost")) {
+            return false;
+        }
+        if (!isNumericHost(host)) {
+            return true;
+        }
+        return isAllowedNumericHost(host);
+    }
+
+    /** 去尾部所有点（`localhost.` / `127.0.0.1.`），再去 IPv6 方括号（`[::1]`）。 */
+    private static String normalizeHost(String rawHost) {
+        if (rawHost == null) {
+            return null;
+        }
+        String host = rawHost.trim().toLowerCase(Locale.ROOT);
+        int end = host.length();
+        while (end > 0 && host.charAt(end - 1) == '.') {
+            end--;
+        }
+        host = host.substring(0, end);
+        if (host.length() >= 2 && host.startsWith("[") && host.endsWith("]")) {
             host = host.substring(1, host.length() - 1);
         }
-        host = host.toLowerCase(Locale.ROOT);
-        if ("localhost".equals(host)) {
-            return false;
+        return host;
+    }
+
+    /** 数字形态：含冒号（IPv6）、{@code 0x} 开头（十六进制整数）、或只含数字与点。 */
+    private static boolean isNumericHost(String host) {
+        if (host.indexOf(':') >= 0) {
+            return true;
         }
-        if (host.startsWith("127.")) {
-            return false;
+        if (host.startsWith("0x")) {
+            return true;
         }
-        if ("::1".equals(host)) {
-            return false;
+        for (int i = 0; i < host.length(); i++) {
+            char c = host.charAt(i);
+            if (!((c >= '0' && c <= '9') || c == '.')) {
+                return false;
+            }
         }
-        if ("0.0.0.0".equals(host)) {
+        return true;
+    }
+
+    /** 数字字面量解析后判地址类别；任何解析失败都拒绝（fail-closed）。 */
+    private static boolean isAllowedNumericHost(String host) {
+        try {
+            InetAddress address;
+            if (host.indexOf(':') >= 0) {
+                if (!IPV6_LITERAL.matcher(host).matches()) {
+                    return false;
+                }
+                address = InetAddress.getByName(host);
+            } else if (host.startsWith("0x")) {
+                address = InetAddress.getByAddress(parseNumericIpv4(host.substring(2), 16));
+            } else if (host.indexOf('.') >= 0) {
+                address = InetAddress.getByAddress(parseDottedIpv4(host));
+            } else {
+                address = InetAddress.getByAddress(parseNumericIpv4(host, 10));
+            }
+            return !isForbiddenAddress(address);
+        } catch (Exception e) {
             return false;
         }
-        return !"169.254.169.254".equals(host);
+    }
+
+    /**
+     * 单个十进制/十六进制整数形式的 IPv4（如 {@code 2130706433} = 127.0.0.1）。
+     * 十进制前导零在 C/OS 里是八进制，语义有歧义 —— 直接拒绝。
+     */
+    private static byte[] parseNumericIpv4(String digits, int radix) {
+        if (digits.isEmpty()
+                || (radix == 10 && digits.length() > 1 && digits.charAt(0) == '0')) {
+            throw new IllegalArgumentException("空的或带前导零的整数");
+        }
+        BigInteger value = new BigInteger(digits, radix);
+        if (value.signum() < 0 || value.bitLength() > 32) {
+            throw new IllegalArgumentException("超出 IPv4 范围");
+        }
+        long v = value.longValue();
+        return new byte[]{(byte) (v >>> 24), (byte) (v >>> 16), (byte) (v >>> 8), (byte) v};
+    }
+
+    /**
+     * 点分 IPv4。只接受规范的四段十进制；段数不为 4、空段、越界或前导零（OS 可能按八进制
+     * 解释，如 {@code 0177.0.0.1}）一律拒绝。
+     */
+    private static byte[] parseDottedIpv4(String host) {
+        String[] parts = host.split("\\.", -1);
+        if (parts.length != 4) {
+            throw new IllegalArgumentException("不是四段 IPv4");
+        }
+        byte[] bytes = new byte[4];
+        for (int i = 0; i < 4; i++) {
+            String part = parts[i];
+            if (part.isEmpty() || part.length() > 3) {
+                throw new IllegalArgumentException("空段或过长的段");
+            }
+            if (part.length() > 1 && part.charAt(0) == '0') {
+                throw new IllegalArgumentException("前导零段语义有歧义");
+            }
+            int value = Integer.parseInt(part);
+            if (value > 255) {
+                throw new IllegalArgumentException("段越界");
+            }
+            bytes[i] = (byte) value;
+        }
+        return bytes;
+    }
+
+    /** 环回 / any-local / link-local；IPv4-mapped IPv6 再按末 4 字节判一次。 */
+    private static boolean isForbiddenAddress(InetAddress address) {
+        if (address.isLoopbackAddress() || address.isAnyLocalAddress() || address.isLinkLocalAddress()) {
+            return true;
+        }
+        byte[] bytes = address.getAddress();
+        if (bytes.length == 4) {
+            return isForbiddenIpv4(bytes, 0);
+        }
+        if (bytes.length == 16 && isIpv4Mapped(bytes)) {
+            return isForbiddenIpv4(bytes, 12);
+        }
+        return false;
+    }
+
+    private static boolean isIpv4Mapped(byte[] bytes) {
+        for (int i = 0; i < 10; i++) {
+            if (bytes[i] != 0) {
+                return false;
+            }
+        }
+        return bytes[10] == (byte) 0xFF && bytes[11] == (byte) 0xFF;
+    }
+
+    /** 0.0.0.0/8（本网络/any）与 127.0.0.0/8 环回、169.254.0.0/16 link-local。 */
+    private static boolean isForbiddenIpv4(byte[] bytes, int offset) {
+        int first = bytes[offset] & 0xFF;
+        int second = bytes[offset + 1] & 0xFF;
+        if (first == 0 || first == 127) {
+            return true;
+        }
+        return first == 169 && second == 254;
     }
 
     /** 订单事件的确定性 notify_id：同一对象同一状态的重试保持同一个 id。 */

java/src/test/java/com/payment/platform/module/payment/unified/service/UnifiedCallbackServiceTest.java

@@ -10,12 +10,14 @@ import com.payment.platform.module.payment.apikey.entity.TenantApiKeyEntity;
 import com.payment.platform.module.payment.apikey.mapper.TenantApiKeyMapper;
 import com.payment.platform.module.payment.openapi.entity.OpenConfEntity;
 import com.payment.platform.module.payment.openapi.mapper.OpenConfMapper;
+import com.payment.platform.module.payment.unified.config.UnifiedHttpConfig;
 import com.payment.platform.module.payment.unified.config.UnifiedPayProperties;
 import com.payment.platform.module.payment.unified.entity.UnifiedOrderEntity;
 import com.payment.platform.module.payment.unified.entity.UnifiedRefundEntity;
 import com.payment.platform.module.payment.unified.enums.UnifiedNotifyStatus;
 import com.payment.platform.module.payment.unified.mapper.UnifiedOrderMapper;
 import com.payment.platform.module.payment.unified.mapper.UnifiedRefundMapper;
+import com.sun.net.httpserver.HttpServer;
 import org.apache.ibatis.builder.MapperBuilderAssistant;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -29,11 +31,16 @@ import org.springframework.http.ResponseEntity;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.client.RestTemplate;
 
+import java.net.InetSocketAddress;
 import java.time.OffsetDateTime;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -365,6 +372,17 @@ class UnifiedCallbackServiceTest {
                 "http://[::1]/x",
                 "http://0.0.0.0/x",
                 "http://169.254.169.254/latest/meta-data",
+                // 等价字面量：字符串比较挡不住，但都能打到本机（评审 C-1 的实测清单）
+                "http://localhost./x",
+                "http://[::ffff:127.0.0.1]/x",
+                "http://[::ffff:7f00:1]/x",
+                "http://[0:0:0:0:0:ffff:7f00:1]/x",
+                "http://[0:0:0:0:0:0:0:1]/x",
+                "http://[::]/x",
+                "http://[::ffff:169.254.169.254]/x",
+                "http://2130706433/x",
+                "http://0127.0.0.1/x",
+                "http://0177.0.0.1/x",
                 "file:///etc/passwd",
                 "javascript:alert(1)",
                 "");
@@ -391,6 +409,45 @@ class UnifiedCallbackServiceTest {
         }
     }
 
+    /**
+     * 端到端对照：用**生产** {@code RestTemplate} + 真实 {@code HttpServer}。三条等价本机
+     * 字面量若通过过滤器就会真的连上（评审 E2E 探针实测 200），所以「0 命中」是行为证据，
+     * 不是对字符串函数的单测；末尾再用 127.0.0.1 正控证明台子本身是通的。
+     */
+    @Test
+    void e2e_hostileEquivalentLiterals_neverEgress() throws Exception {
+        AtomicInteger hits = new AtomicInteger();
+        HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        server.createContext("/", exchange -> {
+            hits.incrementAndGet();
+            exchange.sendResponseHeaders(200, -1);
+            exchange.close();
+        });
+        server.start();
+        int port = server.getAddress().getPort();
+        try {
+            UnifiedPayProperties realProps = new UnifiedPayProperties();
+            realProps.setCallbackTimeoutMs(2000);
+            RestTemplate realRestTemplate = new UnifiedHttpConfig().callbackRestTemplate(realProps);
+            UnifiedCallbackService realService = new UnifiedCallbackService(orderMapper, refundMapper,
+                    apiKeyMapper, confMapper, realRestTemplate, Runnable::run, realProps);
+
+            for (String host : List.of("localhost.", "[::ffff:127.0.0.1]", "2130706433")) {
+                UnifiedOrderEntity order = order("UP1", "2");
+                order.setCallbackUrl("http://" + host + ":" + port + "/x");
+                realService.notifyOrder(order, null);
+            }
+            assertEquals(0, hits.get(), "本机等价字面量必须一次出站都不发生");
+
+            // 正对照：服务器与出站客户端本身是通的，0 命中来自过滤器而不是台子坏了
+            realRestTemplate.postForEntity("http://127.0.0.1:" + port + "/ok",
+                    new HttpEntity<>("x"), String.class);
+            assertEquals(1, hits.get(), "正对照必须命中");
+        } finally {
+            server.stop(0);
+        }
+    }
+
     // ───────────────────────────── 退款事件 ─────────────────────────────
 
     @Test
@@ -443,6 +500,8 @@ class UnifiedCallbackServiceTest {
         assertEquals(UnifiedNotifyStatus.SUCCESS, setValueOf(updates.get(1), "notify_status"));
         assertTrue(whereBoundValues(updates.get(0)).contains("UR1"),
                 "退款状态按 refund_no 更新。实测: " + updates.get(0).getSqlSegment());
+        assertEquals("2", whereValueOf(updates.get(0), "state"),
+                "状态回写必须带事件判别键 state，否则旧事件的 2xx 会覆盖新事件的 FAILED");
     }
 
     @Test
@@ -652,11 +711,15 @@ class UnifiedCallbackServiceTest {
     }
 
     @Test
-    void retryFailed_refundWithOrder_pushesAndMarksSuccess() {
+    void retryFailed_refundWithOrder_pushesAndMarksSuccess() throws Exception {
         UnifiedRefundEntity failed = refund();
         failed.setCallbackUrl("https://merchant.example.com/cb");
         failed.setNotifyStatus(UnifiedNotifyStatus.FAILED);
         failed.setNotifyCount(1);
+        failed.setUpstreamDataJson("""
+                {"refundOrderId":"R1","state":"2",
+                 "channelExtra":{"auth_code":"284211"},"openid":"o-1"}
+                """);
         UnifiedOrderEntity order = order("UP1", "2");
         when(orderMapper.selectList(any())).thenReturn(List.of());
         when(refundMapper.selectList(any())).thenReturn(List.of(failed));
@@ -665,8 +728,13 @@ class UnifiedCallbackServiceTest {
 
         service.retryFailed();
 
-        verify(restTemplate).postForEntity(eq("https://merchant.example.com/cb"),
-                any(HttpEntity.class), eq(String.class));
+        MultiValueMap<String, String> form = formOf(capturePush(restTemplate, "https://merchant.example.com/cb"));
+        JsonNode upstreamNode = contentOf(form).path("upstream");
+        assertEquals("R1", upstreamNode.path("refundOrderId").asText(),
+                "重试报文要用行上的 upstream_data_json 还原上游快照");
+        assertFalse(upstreamNode.has("channelExtra"), "快照里的 channelExtra 仍不得转发");
+        assertFalse(upstreamNode.has("openid"), "快照里的 openid 仍不得转发");
+
         List<LambdaUpdateWrapper<?>> updates = captureRefundUpdates(2);
         assertEquals(UnifiedNotifyStatus.FAILED, setValueOf(updates.get(0), "notify_status"));
         assertEquals(UnifiedNotifyStatus.SUCCESS, setValueOf(updates.get(1), "notify_status"));
@@ -683,6 +751,129 @@ class UnifiedCallbackServiceTest {
         verify(refundMapper).selectList(any());
     }
 
+    @Test
+    void retryFailed_orderUpstreamSnapshot_isRestoredAndWhitelisted() throws Exception {
+        UnifiedOrderEntity failed = order("UP1", "2");
+        failed.setCallbackUrl("https://merchant.example.com/cb");
+        failed.setNotifyStatus(UnifiedNotifyStatus.FAILED);
+        failed.setNotifyCount(1);
+        failed.setUpstreamDataJson("""
+                {"payOrderId":"P1","state":"2","channelExtra":{"auth_code":"284211"},
+                 "openid":"o-1","sign":"DEADBEEF"}
+                """);
+        when(orderMapper.selectList(any())).thenReturn(List.of(failed));
+        when(refundMapper.selectList(any())).thenReturn(List.of());
+        stubPushOk();
+
+        service.retryFailed();
+
+        MultiValueMap<String, String> form = formOf(capturePush(restTemplate, "https://merchant.example.com/cb"));
+        JsonNode upstreamNode = contentOf(form).path("upstream");
+        assertEquals("P1", upstreamNode.path("payOrderId").asText(),
+                "重试报文要用行上的 upstream_data_json 还原上游快照（I-1）");
+        assertEquals("2", upstreamNode.path("state").asText());
+        assertFalse(upstreamNode.has("channelExtra"), "快照里的 channelExtra 仍不得转发");
+        assertFalse(upstreamNode.has("openid"), "快照里的 openid 仍不得转发");
+        assertFalse(upstreamNode.has("sign"), "快照里的 sign 仍不得转发");
+    }
+
+    @Test
+    void retryFailed_invalidUpstreamSnapshot_usesEmptyUpstream() throws Exception {
+        UnifiedOrderEntity failed = order("UP1", "2");
+        failed.setCallbackUrl("https://merchant.example.com/cb");
+        failed.setNotifyStatus(UnifiedNotifyStatus.FAILED);
+        failed.setNotifyCount(1);
+        failed.setUpstreamDataJson("{not-json");
+        when(orderMapper.selectList(any())).thenReturn(List.of(failed));
+        when(refundMapper.selectList(any())).thenReturn(List.of());
+        stubPushOk();
+
+        service.retryFailed();
+
+        MultiValueMap<String, String> form = formOf(capturePush(restTemplate, "https://merchant.example.com/cb"));
+        assertEquals(0, contentOf(form).path("upstream").size(),
+                "非法快照不得让回调失败，upstream 退化为空对象");
+    }
+
+    /**
+     * I-2：同一订单行的 event2（trade_status=2）与 event5（trade_status=5）并发。行上的
+     * trade_status 现在已是 5；event2 的 2xx 晚到时不得把 event5 的 FAILED 覆盖成 SUCCESS，
+     * 否则扫描（只选 FAILED）永远捞不到 event5。
+     */
+    @Test
+    void concurrentEvents_event2SuccessDoesNotOverwriteEvent5Failed() {
+        UnifiedOrderEntity event5 = order("UP1", "5");
+        event5.setCallbackUrl("https://merchant.example.com/cb");
+        UnifiedOrderEntity event2 = order("UP1", "2");
+        event2.setCallbackUrl("https://merchant.example.com/cb");
+
+        // 最小假库：单行，当前 trade_status 已是 5。WHERE 不含事件判别键时按老语义视为命中整行。
+        Map<String, Object> row = new HashMap<>();
+        row.put("trade_status", "5");
+        row.put("notify_status", UnifiedNotifyStatus.FAILED);
+        when(orderMapper.update(isNull(), any())).thenAnswer(invocation -> {
+            LambdaUpdateWrapper<?> wrapper = invocation.getArgument(1);
+            Object whereTradeStatus = whereValueOfOrNull(wrapper, "trade_status");
+            if (whereTradeStatus != null && !Objects.equals(whereTradeStatus, row.get("trade_status"))) {
+                return 0;
+            }
+            row.put("notify_status", setValueOf(wrapper, "notify_status"));
+            return 1;
+        });
+        when(refundMapper.sumOccupiedAmount(ORDER_DB_ID)).thenReturn(30L);
+        when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
+                .thenReturn(new ResponseEntity<>("boom", HttpStatus.INTERNAL_SERVER_ERROR))
+                .thenReturn(new ResponseEntity<>("ok", HttpStatus.OK));
+
+        List<Runnable> pending = new ArrayList<>();
+        UnifiedCallbackService asyncService = new UnifiedCallbackService(orderMapper, refundMapper,
+                apiKeyMapper, confMapper, restTemplate, pending::add, props);
+
+        asyncService.notifyOrder(event5, null);   // FAILED 写入 + 排队
+        asyncService.notifyOrder(event2, null);
+        pending.get(0).run();                      // event5 推送失败，保持 FAILED
+        pending.get(1).run();                      // event2 推送成功，但不得覆盖 event5
+
+        assertEquals(UnifiedNotifyStatus.FAILED, row.get("notify_status"),
+                "event2 的 2xx 不得把 event5 的 FAILED 覆盖成 SUCCESS（否则 event5 永久丢失）");
+        List<LambdaUpdateWrapper<?>> updates = captureOrderUpdates(3);
+        assertEquals("5", whereValueOf(updates.get(0), "trade_status"),
+                "event5 的失败写入必须只命中 trade_status=5 的行");
+        assertEquals("2", whereValueOf(updates.get(2), "trade_status"),
+                "event2 的成功写入必须只命中 trade_status=2 的行");
+    }
+
+    /** I-3：快照构建（sumOccupiedAmount）抛异常时，行不能留在 PENDING —— 扫描只选 FAILED。 */
+    @Test
+    void notifyOrder_contentBuildThrows_recordsFailedWithoutOutbound() {
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setCallbackUrl("https://merchant.example.com/cb");
+        when(refundMapper.sumOccupiedAmount(ORDER_DB_ID)).thenThrow(new RuntimeException("db down"));
+
+        assertDoesNotThrow(() -> service.notifyOrder(order, null));
+
+        verifyNoInteractions(restTemplate);
+        List<LambdaUpdateWrapper<?>> updates = captureOrderUpdates(1);
+        assertEquals(UnifiedNotifyStatus.FAILED, setValueOf(updates.get(0), "notify_status"),
+                "准备阶段异常也必须落 FAILED，否则重试扫描永远捞不到");
+        assertTrue(updates.get(0).getSqlSet().contains("notify_count = notify_count + 1"));
+    }
+
+    /** I-3 的另一半：取址（mapper）抛异常同理。 */
+    @Test
+    void notifyOrder_addressResolutionThrows_recordsFailedWithoutOutbound() {
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setApiKey("K1");
+        when(apiKeyMapper.selectOne(any())).thenThrow(new RuntimeException("db down"));
+
+        assertDoesNotThrow(() -> service.notifyOrder(order, null));
+
+        verifyNoInteractions(restTemplate);
+        List<LambdaUpdateWrapper<?>> updates = captureOrderUpdates(1);
+        assertEquals(UnifiedNotifyStatus.FAILED, setValueOf(updates.get(0), "notify_status"));
+        assertTrue(updates.get(0).getSqlSet().contains("notify_count = notify_count + 1"));
+    }
+
     // ───────────────────────────── 断言工具 ─────────────────────────────
 
     /** 只取 SET 段里某一列的绑定值（SET 与 WHERE 共用参数表，混着找会假绿）。 */
@@ -711,4 +902,22 @@ class UnifiedCallbackServiceTest {
         }
         return out;
     }
+
+    /** 取 WHERE 段某一列的等值绑定；没有这一列就判失败（断言用）。 */
+    private static Object whereValueOf(LambdaUpdateWrapper<?> w, String column) {
+        Object value = whereValueOfOrNull(w, column);
+        assertNotNull(value, "WHERE 里没有这一列: " + column + "，实测: " + w.getSqlSegment());
+        return value;
+    }
+
+    /** 同上，但允许缺失（假库用它模拟「旧实现没有事件判别键」的老语义）。 */
+    private static Object whereValueOfOrNull(LambdaUpdateWrapper<?> w, String column) {
+        Matcher m = Pattern.compile(Pattern.quote(column)
+                        + "\\s*=\\s*#\\{ew\\.paramNameValuePairs\\.([A-Za-z0-9_]+)}")
+                .matcher(w.getSqlSegment());
+        if (!m.find()) {
+            return null;
+        }
+        return w.getParamNameValuePairs().get(m.group(1));
+    }
 }