From b4c6652f669ae2a039659f70ca46c1a0b089eb1a Mon Sep 17 00:00:00 2001 From: yixiong Date: Thu, 9 Jul 2026 09:52:13 +0800 Subject: [PATCH] =?UTF-8?q?fix(review):=20=E4=BB=A3=E7=A0=81=E5=AE=A1?= =?UTF-8?q?=E6=9F=A5=206=20=E9=A1=B9=E4=BF=AE=E5=A4=8D=EF=BC=88M2/M3/m1/m2?= =?UTF-8?q?/m3/m5=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 集中修复 T8/T9/T11/T15 已实现代码的审查问题： - M2 令牌桶 Lua 改用 Redis TIME（消除多节点时钟偏斜） - M3 after 跨租户标记复查授权 + 无授权告警兜底 - m1 before 审计 MsgSeq/MsgRandom null 防御 - m2 HealthService 加 forceDown 开关，实测 DOWN 真返 503 - m3 countByStatus 排除 done（WHERE status IN (0,1,3)） - m5 配额查询异常不缓存（下次重试） - 附带 app.yml console charset GBK→UTF-8 M1 限流审计/事务收窄暂缓。 Co-Authored-By: Claude --- .../java/com/imutil/common/HealthService.java | 17 ++++++++++++- .../java/com/imutil/common/RateLimiter.java | 11 +++++---- .../com/imutil/mapper/DistQueueMapper.java | 2 +- .../service/impl/CallbackServiceImpl.java | 24 ++++++++++++++----- .../com/imutil/tencent/TencentImClient.java | 5 ++-- src/main/resources/app.yml | 2 +- 6 files changed, 45 insertions(+), 16 deletions(-) diff --git a/src/main/java/com/imutil/common/HealthService.java b/src/main/java/com/imutil/common/HealthService.java index 67f2451..538020d 100644 --- a/src/main/java/com/imutil/common/HealthService.java +++ b/src/main/java/com/imutil/common/HealthService.java @@ -37,12 +37,27 @@ public class HealthService { @Inject private JedisPool jedisPool; + /** 运维测试用：强制返回 DOWN，验证 LB/k8s 探针摘流（默认关） */ + @Inject("${imutil.health.forceDown:false}") + private boolean forceDown; + /** * 采集健康指标 * * @return 含 status / pg / redis / distQueue / timestamp */ public Map health() { + // 运维测试开关：强制 DOWN，验证探针据 503 摘流 + if (forceDown) { + Map r = new LinkedHashMap<>(); + r.put("status", "DOWN"); + r.put("pg", "DOWN"); + r.put("redis", "UP"); + r.put("distQueue", new LinkedHashMap<>()); + r.put("forced", true); + r.put("timestamp", OffsetDateTime.now().toString()); + return r; + } Map distQueue = new LinkedHashMap<>(); boolean pgUp; try { @@ -58,8 +73,8 @@ public class HealthService { } distQueue.put("pending", cnt[0]); distQueue.put("processing", cnt[1]); - distQueue.put("done", cnt[2]); distQueue.put("dead", cnt[3]); + // done(status=2) 不统计：无索引且无限增长，健康检查只关心堆积/死信 pgUp = true; } catch (Exception e) { log.warn("健康检查 PG 查询失败 : {}", e.getMessage()); diff --git a/src/main/java/com/imutil/common/RateLimiter.java b/src/main/java/com/imutil/common/RateLimiter.java index 6414117..efaf357 100644 --- a/src/main/java/com/imutil/common/RateLimiter.java +++ b/src/main/java/com/imutil/common/RateLimiter.java @@ -24,13 +24,15 @@ import java.util.List; @Component public class RateLimiter { - /** 令牌桶 Lua：按时间填充令牌、扣减，返回 1=放行 0=超限 */ + // 令牌桶 Lua：用 Redis 服务端 TIME（避免多节点客户端时钟偏斜导致令牌凭空填充）， + // 按时间填充令牌、扣减，返回 1=放行 0=超限 private static final String TOKEN_BUCKET_LUA = "local key = KEYS[1] " + "local capacity = tonumber(ARGV[1]) " + "local rate = tonumber(ARGV[2]) " + - "local now = tonumber(ARGV[3]) " + - "local requested = tonumber(ARGV[4]) " + + "local requested = tonumber(ARGV[3]) " + + "local tt = redis.call('TIME') " + + "local now = tonumber(tt[1]) * 1000 + math.floor(tonumber(tt[2]) / 1000) " + "local tokens = tonumber(redis.call('HGET', key, 'tokens')) " + "local last = tonumber(redis.call('HGET', key, 'last')) " + "if tokens == nil then tokens = capacity end " + @@ -82,10 +84,9 @@ public class RateLimiter { return true; } try { - long now = System.currentTimeMillis(); Object r = redisService.eval(TOKEN_BUCKET_LUA, List.of(key), - List.of(String.valueOf(qps), String.valueOf(qps), String.valueOf(now), "1")); + List.of(String.valueOf(qps), String.valueOf(qps), "1")); // Lua 返回 number，Jedis 转 Long long allowed = (r instanceof Number) ? ((Number) r).longValue() : Long.parseLong(String.valueOf(r)); return allowed == 1L; diff --git a/src/main/java/com/imutil/mapper/DistQueueMapper.java b/src/main/java/com/imutil/mapper/DistQueueMapper.java index cb13c2a..90bcac7 100644 --- a/src/main/java/com/imutil/mapper/DistQueueMapper.java +++ b/src/main/java/com/imutil/mapper/DistQueueMapper.java @@ -84,6 +84,6 @@ public interface DistQueueMapper extends BaseMapper { * * @return 每项含 status（0=pending/1=processing/2=done/3=dead）与 cnt */ - @org.apache.ibatis.annotations.Select("SELECT status, count(*) AS cnt FROM dist_queue GROUP BY status") + @org.apache.ibatis.annotations.Select("SELECT status, count(*) AS cnt FROM dist_queue WHERE status IN (0, 1, 3) GROUP BY status") java.util.List> countByStatus(); } diff --git a/src/main/java/com/imutil/service/impl/CallbackServiceImpl.java b/src/main/java/com/imutil/service/impl/CallbackServiceImpl.java index 972fcfc..8c68427 100644 --- a/src/main/java/com/imutil/service/impl/CallbackServiceImpl.java +++ b/src/main/java/com/imutil/service/impl/CallbackServiceImpl.java @@ -96,12 +96,18 @@ public class CallbackServiceImpl implements CallbackService { ImMessage msg = parseMessage(callbackCommand, node, tenantId); if (msg != null) { msgKey = msg.getMsgKey(); - // 跨租户标记：能到达 after 的跨租户 C2C 消息必经 before 放行（有授权）； - // 未授权的已被 before 拦截（腾讯不投递、不触发 after），故 after 见到跨租户即标记。 + // 跨租户标记：复查授权确认（before 已拦截未授权）。 + // 若 before 未开启/失效，未授权消息会漏到 after，此处查不到授权即告警（兜底发现隔离异常）。 if (msg.getConvType() != null && msg.getConvType() == 1) { String toTenant = parsePrefix(msg.getConvId()); if (toTenant != null && !toTenant.equals(tenantId)) { - msg.setIsCrossTenant(true); + if (crossTenantService.checkSendMsgGrant( + tenantId, msg.getFromAccount(), toTenant, msg.getConvId()) != null) { + msg.setIsCrossTenant(true); + } else { + log.warn("疑似未授权跨租户消息到达 after（before 可能未开启/失效）from={} to={}", + msg.getFromAccount(), msg.getConvId()); + } } } long exists = imMessageMapper.selectCount(Wrappers.lambdaQuery() @@ -150,9 +156,15 @@ public class CallbackServiceImpl implements CallbackService { return fail(); } // 命中授权：放行 + 写审计（before 阶段记录放行决策，msgKey 便于追溯） - long msgSeq = node.get("MsgSeq").getLong(); - long msgRandom = node.get("MsgRandom").getLong(); - String msgKey = MsgKeys.build(from, to, msgSeq, msgRandom); + // MsgSeq/MsgRandom 缺失时审计 msgKey 留空，不影响拦截决策（授权检查已完成） + String msgKey = null; + try { + long msgSeq = node.get("MsgSeq").getLong(); + long msgRandom = node.get("MsgRandom").getLong(); + msgKey = MsgKeys.build(from, to, msgSeq, msgRandom); + } catch (Exception ignore) { + // 回调体缺字段无法算 msgKey，审计仍写（msgKey=null） + } crossTenantService.audit(grantId, msgKey, from, to); log.info("跨租户授权放行 from={} to={} grant={}", from, to, grantId); return ok(); diff --git a/src/main/java/com/imutil/tencent/TencentImClient.java b/src/main/java/com/imutil/tencent/TencentImClient.java index 18fc1ce..dad45f1 100644 --- a/src/main/java/com/imutil/tencent/TencentImClient.java +++ b/src/main/java/com/imutil/tencent/TencentImClient.java @@ -156,10 +156,11 @@ public class TencentImClient { if (t != null && t.getQuotaImQps() != null && t.getQuotaImQps() > 0) { qps = t.getQuotaImQps(); } + // 仅查询成功才缓存（含"租户无配额配置→用默认"）；异常时不缓存，让下次重试 + localCache.put(cacheKey, String.valueOf(qps)); } catch (Exception e) { - log.warn("查询租户配额失败 tenant={}，用默认 {} : {}", tid, defaultImQps, e.getMessage()); + log.warn("查询租户配额失败 tenant={}，用默认 {}（不缓存，下次重试）: {}", tid, defaultImQps, e.getMessage()); } - localCache.put(cacheKey, String.valueOf(qps)); return qps; } diff --git a/src/main/resources/app.yml b/src/main/resources/app.yml index bcc826f..c1030ab 100644 --- a/src/main/resources/app.yml +++ b/src/main/resources/app.yml @@ -12,7 +12,7 @@ solon.logging.logger: level: INFO solon.logging.appender: console: - charset: GBK + charset: UTF-8 pattern: "%d{yyyy-MM-dd HH:mm:ss.SSS} %highlight(%-5level) %magenta(${PID:-}) --- %-15([%15.15thread]) %-56(%cyan(%-40.40logger{39}%L)) : %msg%n" file: charset: UTF-8