fix: scope realtime fanout dedup per node

2026-09-14 20:15:22 +08:00
parent 6291fa2b60
commit 79babdb6aa
2 changed files with 22 additions and 1 deletions
@@ -851,7 +851,7 @@ func (r *redisGroupMessageRuntime) StartRealtimeFanoutSubscriber(ctx context.Con
 					continue
 				}
 				if envelope.TenantID.IsValid() && envelope.DedupKey != "" {
-					id := fmt.Sprintf("%d:%s", envelope.TenantID, envelope.DedupKey)
+					id := realtimeFanoutDedupClaimID(envelope.TenantID, nodeID, envelope.DedupKey)
 					claimed, err := r.ClaimShortState("rt_dedup", id, 5*time.Minute)
 					if err != nil || !claimed {
 						continue
@@ -1110,6 +1110,13 @@ func (r *redisGroupMessageRuntime) realtimeFanoutPattern() string {
 	return strings.TrimSuffix(r.prefix, ":") + ":*:rt:fanout"
 }

+func realtimeFanoutDedupClaimID(tenantID types.TenantID, nodeID, dedupKey string) string {
+	if nodeID == "" {
+		nodeID = "local"
+	}
+	return fmt.Sprintf("%d:%s:%s", tenantID, nodeID, dedupKey)
+}
+
 func (r *redisGroupMessageRuntime) objectCacheKey(tenantID types.TenantID, namespace, id string) string {
 	return r.key(tenantID, "cache", namespace, id)
 }