fix(review): 代码审查 6 项修复(M2/M3/m1/m2/m3/m5)
集中修复 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 <noreply@anthropic.com>
This commit is contained in:
@@ -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<String, Object> health() {
|
||||
// 运维测试开关:强制 DOWN,验证探针据 503 摘流
|
||||
if (forceDown) {
|
||||
Map<String, Object> 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<String, Object> 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());
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -84,6 +84,6 @@ public interface DistQueueMapper extends BaseMapper<DistQueue> {
|
||||
*
|
||||
* @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<java.util.Map<String, Object>> countByStatus();
|
||||
}
|
||||
|
||||
@@ -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)) {
|
||||
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.<ImMessage>lambdaQuery()
|
||||
@@ -150,9 +156,15 @@ public class CallbackServiceImpl implements CallbackService {
|
||||
return fail();
|
||||
}
|
||||
// 命中授权:放行 + 写审计(before 阶段记录放行决策,msgKey 便于追溯)
|
||||
// MsgSeq/MsgRandom 缺失时审计 msgKey 留空,不影响拦截决策(授权检查已完成)
|
||||
String msgKey = null;
|
||||
try {
|
||||
long msgSeq = node.get("MsgSeq").getLong();
|
||||
long msgRandom = node.get("MsgRandom").getLong();
|
||||
String msgKey = MsgKeys.build(from, to, msgSeq, msgRandom);
|
||||
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();
|
||||
|
||||
@@ -156,10 +156,11 @@ public class TencentImClient {
|
||||
if (t != null && t.getQuotaImQps() != null && t.getQuotaImQps() > 0) {
|
||||
qps = t.getQuotaImQps();
|
||||
}
|
||||
} catch (Exception e) {
|
||||
log.warn("查询租户配额失败 tenant={},用默认 {} : {}", tid, defaultImQps, e.getMessage());
|
||||
}
|
||||
// 仅查询成功才缓存(含"租户无配额配置→用默认");异常时不缓存,让下次重试
|
||||
localCache.put(cacheKey, String.valueOf(qps));
|
||||
} catch (Exception e) {
|
||||
log.warn("查询租户配额失败 tenant={},用默认 {}(不缓存,下次重试): {}", tid, defaultImQps, e.getMessage());
|
||||
}
|
||||
return qps;
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user